The Consumer API

We will continue from where we left off in the previous chapter. In this chapter, we will focus on consuming the same Notification messages in our application. As discussed in Chapter 2, a Consumer is an application that reads data from Kafka topics by subscribing to them.

Consumers maintain connectivity to the Kafka cluster using the concept of heartbeat. This heartbeat allows the Zookeeper or Broker coordinator to be aware of whether the consumer is constantly connected to the cluster or not. In the absence of a heartbeat, the broker coordinator knows that the consumer is no longer connected and needs to rebalance the load among other consumers.

As discussed in Chapter 2, consumers are also grouped into Consumer Groups so that they can share the partitions of the topics they subscribe to. We have already discussed the different strategies that can be adapted based on various use-cases in Chapter 2. So let's try the practical implementation of the same here.

Kafka Consumer Implementation

Let's create a simple Maven-based Java application in any of our favorite IDEs. We will add the following libraries as part of the initial POM:

    <dependencies>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.13</artifactId>
            <version>3.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.36</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.13.1</version>
        </dependency>
        <dependency>
            <groupId>com.github.javafaker</groupId>
            <artifactId>javafaker</artifactId>
            <version>1.0.2</version>
        </dependency>
    </dependencies>