Go Language

Write Kafka consumer with Golang and Sarama

Lets write a kafka consumer in golang using sarama that consumes messages from a topic

Golang Community Moderator. Software Engineer - Open Source Enthusiast. Goper Lover

Published · Oct 19 2019

As promised, it's time to see Apache Kafka in action with a couple of microservices written in Go. You will be amazed how easy it's to have it up and running.

Meet Sarama
“Sarama is an MIT-licensed Go client library for Apache Kafka version 0.8 (and later)”. That’s from the Readme on Sarama github page (and arguable for a beginner, that might be the only thing that makes sense at first there).

Aside being robust, that Sarama implements the recent version of Kafka makes it the golang library of choice to use with Kafka. The consumer groups feature in recent Kafka versions finds great use in horizontally scaled applications because it ensures that for replicated consumers, duplicate execution of published message doesn’t occur.
. . .

Golang Consumer
There is a Kafka library for golang call sarama. I’m using sarama library to build the producer. Read more about sarama library from here.

Consumer reads messages from topic senz. You can take whatever action with the read messages(for an example index message inelasticserarch). To publish messages tosenztopic I have written a producer with golang. Please read about it from Kafka producer with golang.

Consumer group is a set of consumers which has a unique group id. Each consumer group is a subscriber to one or more kafka topics. Each consumer group maintains its offset per topic partition. 

A record gets delivered to only one consumer in a consumer group. Each consumer in a consumer group processes records and only one consumer in that group will get the same record. Consumers in a consumer group load balance record processing.

This is slightly more complicated, as we want to have our consumer to be a member of a consumer group.

Following is the code of the consumer.

Code is simple and self explanatory. What I’m doing here is initializing ConsumerGroup for given topic senz. Then starts to consume messages from that consumer group.  Messages will be available via ConsumerGroup.Messages() channel.

It's important to commit to the ZooKeeper the fact that the message is read, because by default it is nod done and you might find yourself reading the same message multiple times after restarts.
err:= cg.CommitUpto(msg)

Joining consumer group
cg, err:= consumergroup.JoinConsumerGroup(cgroup, []string{topic}, []string{zookeeperConn}, config)

Run consumer
go run consumer.go

When publishing messages to kafka topic senz those messages will be consumed by the conzumer and log in to stdout.