kafka_sarama_sync_producer_send_message_with_partition_offset_logging.go

1package main
2
3import (
4	"fmt"
5	"log"
6
7	"github.com/IBM/sarama"
8)
9
10func main() {
11	// Setup configuration
12	config := sarama.NewConfig()
13	config.Producer.Return.Successes = true
14	config.Producer.RequiredAcks = sarama.WaitForAll
15	config.Producer.Retry.Max = 5
16
17	// Define brokers and topic
18	brokers := []string{"localhost:9092"}
19	topic := "important_topic"
20
21	// Create a new SyncProducer
22	producer, err := sarama.NewSyncProducer(brokers, config)
23	if err != nil {
24		log.Panic(err)
25	}
26
27	defer func() {
28		if err := producer.Close(); err != nil {
29			log.Panic(err)
30		}
31	}()
32
33	// Define the message
34	msg := &sarama.ProducerMessage{
35		Topic: topic,
36		Value: sarama.StringEncoder("Something cool happened"),
37	}
38
39	// Send the message
40	partition, offset, err := producer.SendMessage(msg)
41	if err != nil {
42		log.Printf("FAILED to send message: %s\n", err)
43	} else {
44		fmt.Printf("Message is stored in topic(%s)/partition(%d)/offset(%d)\n", topic, partition, offset)
45	}
46}