How to create a kafka consumer group in Golang?

Question

An available library is sarama (or its expansion sarama-cluster) however no consumer group example are provided, not in sarama nor in sarama-cluster.

I do not understand the API. May I have an example of creating a consumer group for a topic?

Snehal Kumar · Accepted Answer

The consumer group is specified by the second argument of the cluster consumer "constructor". Here's a very basic sketch: import ( "github.com/Shopify/sarama" "github.com/bsm/sarama-cluster" ) conf := cluster.NewConfig() // add config values brokers := []string{"kafka-1:9092", "kafka-2:9092"} group := "Your-Consumer-Group" topics := []{"topicName"} consumer := cluster.NewConsumer(broker, group, topics, conf) And so you'll have a consumer belonging to the specified consumer group.

Prasad Patil · Answer

There is no need to use sarama-cluster library. It is DEPRECATED for apache kafka integration. Sarama original library itself provide a way to connect to kafka cluster using consumer group. We need to create client and then we initialize consumer group where we create claims and wait for message channel to receive message. Initializing client :- kfversion, err := sarama.ParseKafkaVersion(kafkaVersion) // kafkaVersion is the version of kafka server like 0.11.0.2 if err != nil { log.Println(err) } config := sarama.NewConfig() config.Version = kfversion config.Consumer.Return.Errors = true // Start with a client client, err := sarama.NewClient([]string{brokerAddr}, config) if err != nil { log.Println(err) } defer func() { _ = client.Close() }() Connection to consumer group :- // Start a new consumer group group, err := sarama.NewConsumerGroupFromClient(consumer_group, client) if err != nil { log.Println(err) } defer func() { _ = group.Close() }() Start consuming messages from topic partition :- // Iterate over consumer sessions. ctx := context.Background() for { topics := []string{topicName} handler := &Message{} err := group.Consume(ctx, topics, handler) if err != nil { log.Println(err) } }

Go Language

Similar Questions

How to copy one array to another in Go?

How to turn []interface{} into arguments to a non-variadic function?

How do I SET the GOPATH environment variable?

On a mission to build Next-Gen Community Platform for Developers

Facebook

Twitter

Linkedin

Instagram