Go Language

Golang: Implementing kafka Consumers & Producers using sarama

In this post I will tell you briefly about what is Kafka. How to create and consume messages using sarama

Published · Jul 28 2019

  
In this post I will tell you briefly about what is Kafka, we will learn how to write kafka producer & consumer ingolang  So, let’s start with what is Apache Kafka itself.
. . .

What Is Apache Kafka?
Apache Kafka is an open-source stream processing software platform which started out at Linkedin. It’s written in Scala and Java. 

Kafka is based on commit log, which means Kafka stores a log of records and it will keep a track of what’s happening. This commit log is similar with common RDBMS uses.

 Kafka Architecture. Source: http://kth.diva-portal.org/smash/get/diva2:813137/FULLTEXT01.pdf 

To make it easier to imagine, it’s more like a queue where you always append a new data into the tail.
  

How Kafka stores a log, image source: https://www.confluent.io/blog/okay-store-data-apache-kafka/While it seems simple, but in Kafka it can maintain the records into several partitions with same topic. 

Based on Kafka documentation’s, a topic is a category or feed name to which records are published. So, rather than just write into one queue like the image above, Kafka can writes into several queue with same topic name.

  
For each topic, the Kafka cluster maintains a partitioned log that looks like this. Source: https://kafka.apache.org/documentation/#designImagine a topic is a name of a hotel. 

And imagine that the partition topic is like that hotel have 3 lift elevators with same direction, let’s say: up direction. In a busy day, when a guest of a hotel want to go to upstairs, they will choose one of three elevators which have a few passengers on it. 

With same analogical logic, when a data arrives, we can tell Kafka to write into specific partition.
. . .

Kafka Consumer Group
It’s all about how Kafka handle an incoming data. Now, let’s see how we can read or consume the data from Kafka.

In Kafka you can consume data from specific partition of a topic, or you can consume it from all partition. Interestingly, you can subscribe the data using several clients/workers and make each of it retrieve different data from different partition using consumer group. So, what is a consumer group?

Consumer group is like a label of a group of consumer. For each consumer under the same label, you will consume different messages. Hmm okay, in simpler way, it’s like when you are at school, your teacher wants you to make a group of 3 persons. 

Then, your teacher will give each group a label of name, for example group 1 given a name “Tiger”, while the other one “Apple”, and so on. From now on, you and your other 2 friends are recognized as one entity rather than 3 entity.

  
Kafka is usually compared to a queuing system such as RabbitMQ. What makes the difference is that after consuming the log, Kafka doesn’t delete it. In that way, messages stay in Kafka longer, and they can be replayed.
. . .
Setting Up Kafka
This tutorial assumes you are starting fresh and have no existing Kafka or ZooKeeper data. Since Kafka console scripts are different for Unix-based and Windows platforms, on Windows platforms use bin\windows\ instead of bin/, and change the script extension to .bat.

Step 1: Download the code
Download the 2.4.1 release and un-tar it.
tar -xzf kafka_2.12-2.4.1.tgz
cd kafka_2.12-2.4.1


Step 2: Start the server
Kafka uses ZooKeeper so you need to first start a ZooKeeper server if you don't already have one. You can use the convenience script packaged with kafka to get a quick-and-dirty single-node ZooKeeper instance.
bin/zookeeper-server-start.sh config/zookeeper.properties

Now start the Kafka server:
bin/kafka-server-start.sh config/server.properties

Step 3: Create a topic
Let's create a topic named "test" with a single partition and only one replica:
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

We can now see that topic if we run the list topic command:
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

#output
test

Alternatively, instead of manually creating topics you can also configure your brokers to auto-create topics when a non-existent topic is published to.

Step 4: Send some messages
Kafka comes with a command line client that will take input from a file or from standard input and send it out as messages to the Kafka cluster. By default, each line will be sent as a separate message.

Run the producer and then type a few messages into the console to send to the server.
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

#send messages to test topic
This is a message
This is another message


Step 5: Start a consumer
Kafka also has a command line consumer that will dump out messages to standard output.
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

#Received messages
This is a message
This is another message

If you have each of the above commands running in a different terminal then you should now be able to type messages into the producer terminal and see them appear in the consumer terminal.

All of the command line tools have additional options; running the command with no arguments will display usage information documenting them in more detail.

. . .
Writing a Producer in Go
Sarama has two kinds of producers, the SyncProducer and AsyncProducer. As you might have guessed, the SyncProducer publishes messages, blocking until they have been acknowledged while the AsyncProducer publishes messages using a non-blocking API. The choice will largely depend on the design of your overall architecture. However you should note that
The SyncProducer is a very thin wrapper around the AsyncProducer — the SendMessage method simply pushes the message onto the underlying async queue, then waits for the response. As such, you should get effectively identical performance with either.

Here’s a sample code snippet to create a Sarama producer:
To create an AsyncProducer we have instead
producer, err := sarama.NewAsyncProducer(brokers, config)

Check this out for more producer configuration options. Here is the complete code for producer:

For all available ProducerMessage configuration, including how to specify the partition to send message to using partition key, checkout the documentation.

SendMessage returns the partition the message was sent to, the offset of the message on Kafka and error information.

Note: be sure to shutdown the producer to avoid memory leaks by calling the Close() function on the producer instance.
. . .
Writing a Consumer in Go
Like the producer, you can provide a sarama config struct as second argument with your desired configurations.

To create a consumer, you will need addresses of the Kafka brokers
To setup the consumer to listen for messages, we provide the topic, partitions and offset the consumer is to subscribe to. It can subscribe to more than one partition and topic. To have a consumer subscribe to all partitions on a topic, use the partitions function on the consumer to return all available partitions.

Once a consumer is subscribed to a topic, sarama returns a PartitonConsumer. The PartitionConsumer processes Kafka messages from the given topic and partition.

The simplest way of using a PartitionConsumer is to loop over its messages channel using a for/range loop. The PartitionConsumer will only stop itself in one case: when the offset being consumed is reported as out of range by the brokers. In this case you should decide what you want to do (try a different offset, notify a human, etc) and handle it appropriately.

For subscriptions to multiple topics/partitions, leverage goroutines and channels to range over and retrieve each PartitionConsumers messages.

Like the producer, ensure you call AsyncClose() on the PartitionConsumer once you are done. Here is complete code for consumer:


. . .
Conclusion
In this tutorial, we have learned how to create kafka consumer and kafka producer in golang with sarama lib.

If you have any questions or comments, feel free to leave them in the section below.