Connect with Kafka brokers using Ballerina.

This blog is about how you can write Kafka consumers and producers using Ballerina. If you are interested in more exploring the Ballerina language please visit official site [1] for more information. Apache Kafka is popular messaging system. If you would like explore more on Apache Kafka, some fundamental knowledge messaging systems, queuing and publish-subscribe messaging, please read this introduction Apache Kafka. [2] Section “Kafka as a Messaging System” provides a great overview of Kafka as opposed to traditional messaging systems.

In summary, what we are going to do here is, we will setup single Kafka broker cluster, then we will write a Ballerina service to consume a Kafka topic from remote Kafka cluster, then will write Ballerina main program to produce messages to that same topic, so that we could observe produced messages are consumed from created service.

Sources for Ballerina Kafka connector is available on [3].

Build Kafka Connector from sources, extract ballerina-kafka-connector-<version>.zip and copy containing jars in to Ballerina runtime <BRE_HOME>/bre/lib/ directory. You can find latest runtime downloads from here. [4]

Download Kafka 1.0.0 version from here. [5] Extract contents of kafka_2.12–1.0.0.tgz.

Start ZooKeeper instance using default configuration.

kafka_2.12–1.0.0$ bin/zookeeper-server-start.sh config/zookeeper.properties

Start single Kafka broker instance using default configuration.

kafka_2.12–1.0.0$ bin/kafka-server-start.sh config/server.properties

Here we create a Kafka server on host : localhost port : 9092.

Now we have setup a working Kafka cluster. Let’s create a topic on Kafka cluster.

kafka_2.12–1.0.0$ bin/kafka-topics.sh --create --topic test-topic --zookeeper localhost:2181 --replication-factor 1 --partitions 2

Here we create test-topic consists of two partition and single replication factor for that partitions.

Ballerina sources (.bal files) for consumer/producer samples are taken from Kafka ballerina-by-example repository. [6]

Let’s start a Ballerina service which consumes Kafka records from test-topic.

ballerina-0.95.6/bin$ ./ballerina run ../location/kafka-message-consumer-service-advanced.bal 

Code for Kafka consumer service.

Here we basically have started kafka-message-consumer-service-advanced sample consumer service. The consumer service code is pretty straightforward. Kafka consumer configurations are provided as service level annotations. We basically provides hosts /ports for the remote brokers connect to, subscribing topic, Kafka consumer group id, Kafka consumer poll cycle interval and most importantly you may have noticed we have enabled manual offset commit ( disabled auto offsets commits ) for the consumer. Basically what that means is, we need to make manual offsets commits programmatically from the program itself. You could notice the line consumer.commit(); where we basically commit all the offsets which consumer has consumed up to that point, once after we have processed all the dispatched records to service.

Let’s start a Ballerina service which produces a Kafka record to test-topic.

ballerina-0.95.6/bin$ ./ballerina run ../location/kafka-message-producer-advanced.bal 

Code for Kafka producer main program.

Here we basically have started kafka-message-producer-advanced sample main program. As you go through the producer code, it’s pretty straightforward. Here we basically create a Kafka producer client which is capable of producing messages to Kafka broker on host / port provided in producer configuration. Using that Kafka producer client, we basically publish a string message to topic test-topic partition 1.

Once we run the producer main program, we could observe the following logs from the console for consumer Ballerina service.

Topic: test-topic Received Message: Hello World Advanced

That means we have received the message we have published via the producer client.

Here you have it. I think there’s lot room to play with the samples provided. We are currently working towards releasing the first version for Kafka connector. I will be writing more on Kafka related material covering wide variety aspects in future. Stay tuned !

[1] https://ballerinalang.org/

[2] https://kafka.apache.org/intro

[3] https://github.com/djkevincr/connector-kafka

[4] https://ballerinalang.org/downloads/

[5] https://kafka.apache.org/downloads

[6] https://github.com/djkevincr/connector-kafka/tree/master/samples/ballerina-by-example/examples