Complete Kafka Consumer with Java

In IntelliJ IDEA, open the project created in the last section.

These are the steps to create a Kafka consumer:

1. Create a Java Class ConsumerDemo.java

2. Create the consumer properties

3. Create a consumer

4. Subscribe the consumer to a specific topic

5. Create a Poll loop to receive data

Create it as io.conduktor.demos.kafka.ConsumerDemo and insert the following content:

Similar to the producer properties, Apache Kafka offers various different properties for creating a consumer as well. To know about each consumer property, visit https://kafka.apache.org/documentation/#consumerconfigs Apache Kafka > Documentation > Configuration > Consumer Configs.

Here, we will list the required properties of a consumer:

• key.deserializer: It is a Deserializer class for the key, which is used to implement the org.apache.kafka.common.serialization.Deserializer interface.

• value.deserializer: A Deserializer class for value which implements the org.apache.kafka.common.serialization.Deserializer interface.

• bootstrap.servers: It is a list of host and port pairs that are used to establish an initial connection with the Kafka cluster. It does not contain a full set of servers that a client requires. Only the servers which are required for bootstrapping are required.

• group.id: It is a unique string that identifies the consumer of a consumer group.

• auto.offset.reset: This property is required when no initial offset is present or if the current offset does not exist anymore on the server. There are the following values used to reset the offset values:

  • earliest: This offset variable automatically reset the value to its earliest offset.

  • latest: This offset variable reset the offset value to its latest offset.

  • none: If no previous offset is found for the previous group, it throws an exception to the consumer.

These are some essential properties that are required to implement a consumer. Let's implement using IntelliJ IDEA.

Create an object of KafkaConsumer leveraging our properties, as shown below:

To read the messages from a topic, we need to connect the consumer to the specified topic. Here, we use Arrays.asList(),as it allows our consumer to subscribe to multiple topics.

Below code shows the implementation of subscription of the consumer to one topic:

The consumer reads data from Kafka through the polling method. The poll method returns the data that hasn't been fetched yet by the consumer subscribed to the partitions. The duration of the poll call for example .poll(Duration.ofMillis(100)) is the amount of time to block on this call before returning an empty list in case no data was returned (also called long polling)

Run the Java application.

It will run displaying messages from the topic, demo_java, created in the last section.

Consumer application runs in an infinite loop, use IntelliJ stop button to stop your application.

We have seen that consumers can share reads in a Consumer Group in order to scale. We can achieve this using the CLI, and of course using Java (right below)

In the last section, we assigned a group id of my-fourth-application to the consumer.

Stop the consumer application, and re-run it. You will notice there are no messages displayed in the console. This is due to the fact that Kafka keeps track of consumer reads (consumer groups offsets) and therefore the consumer will not encounter the same messages twice.

If you change the group id to my-fifth-application as shown below.

Re-start the application. It will display all the messages in the topic demo_java from the beginning.

Moving partition ownership from one consumer to another is called a rebalance. Rebalances are important because they provide the consumer group with high availability and scalability.

Currently, we only have one consumer in our group, and therefore that consumer reads from all the topic partitions.

To see rebalances in action, launch another consumer by running the same consumer application.

Make sure you have configured to run multiple instances of the application in project setting:

Then, edit your configuration to allow for Multiple Instances

Then apply and okay to save your changes.

Now you can run your consumer multiples times.

The first consumer displays the following information:

As we see, the consumer is reading the latest consumer offsets and is assigned to the three partitions in the topic demo java Assignment(partitions=[demo_java-0, demo_java-1, demo_java-2])

Launch a second consumer using the same process defined as above (do not stop the previous one) and notice the rebalance:

Second Consumer Log

As you can see partitions demo_java-0, demo_java-1 have been assigned to the consumer

First Consumer Log

As you can see, the partitions assignment was revoked and after the rebalance, only demo_java-2 was assigned to this consumer .

Produce a few messages on the topic using the Kafka producer application that we developed in the last section. The two consumer applications will receive messages from their respective partitions only.

Using the Kafka Consumer Java API, offsets are committed regularly and automatically in order to enable at-least-once reading scenarios.

You can get a refresher on Consumer Offsets here.

By default, the property enable.auto.commit=true and therefore offsets are committed automatically with a frequency controlled by the config auto.commit.interval.ms.

The process of committing the offsets happens when the .poll() function is called and the time between two calls to .poll() is greater than the setting auto.commit.interval.ms (5 seconds by default).

This means that to be in an "at-least-once" processing use case (the most desirable one), you need to ensure all the messages in your consumer code are successfully processed before performing another .poll() call (which is the case in the sample code defined above). If this is not the case, then offsets could be committed before the messages are actually processed, therefore resulting in an "at-most once" processing pattern, possibly resulting in message skipping (which is undesirable).

In that (rare) case, you must disable enable.auto.commit, and most likely most processing to a separate thread, and then from time to time call .commitSync() or .commitAsync()with the correct offsets manually.

This complicated use case is discussed in the Kafka Consumer Documentation under the section "Automatic Offset Committing".

Currently our consumer is running an infinite loop with while(true) but we can catch an Exception that happens when our consumer is shutting down.

For this, we need to call consumer.wakeup() which will trigger a WakeupException next time the .poll() function is called.

The WakeupException itself does not need to be handled, but then in a finally{} block we can call consumer.close() which will take care of:

• committing the offsets if needed

• close the connection to Kafka

In order to call consumer.wakeup() we need to use a ShutdownHook. That ShutdownHook needs to be linked to the main thread in order to wait for all threads to complete before shutting down the program.

The complete code is seen below:

The important part to look at is to have a try around our while loop and catch for the WakeupException that we will trigger from within the loop.

All in all, the output with a graceful shutdown is the following: