Apache Kafka – Create Consumer using Java

Kafka Consumer is used to reading data from a topic and remember a topic again is identified by its name. So the consumers are smart enough and they will know which broker to read from and which partitions to read from. And in case of broker failures, the consumers know how to recover and this is again a good property of Apache Kafka. Now data for the consumers is going to be read in order within each partition. In this article, we are going to discuss the step-by-step implementation of how to Create an Apache Kafka Consumer using Java.

Step by Step Implementation

Step 1: Create a New Apache Kafka Project in IntelliJ

To create a new Apache Kafka Project in IntelliJ using Java and Maven please refer to How to Create an Apache Kafka Project in IntelliJ using Java and Maven.

Step 2: Install and Run Apache Kafka

To Install and Run Apache Kafka in your local system please refer to How to Install and Run Apache Kafka.

Step 3: Create a Consumer

First, we have to create Consumer Properties. And to create Consumer Properties refer to the below code snippet

Create Consumer Properties:

String bootstrapServer = "127.0.0.1:9092";
String groupId = "gfg-consumer-group";


// Create Consumer Properties
Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

Create Kafka Consumer:

KafkaConsumer<String, String> consumer =
                new KafkaConsumer<>(properties);

Subscribe Consumer to Our Topics:

String topics = "gfg_topic";
consumer.subscribe(List.of(topics));

Note: You can consume from a list of topics.

Poll the data:

while (true) {

   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

   for (ConsumerRecord<String, String> record : records) {
                logger.info("Key: " + record.key() +
                            " Value: " + record.value() +
                            " Partition: " + record.partition() +
                            " Offset: " + record.offset()
               );
       }
}

Below is the complete code. Comments are added inside the code to understand the code in more detail.

Step 4: Run the Application

Now run the application and below is the output.

"C:\Users\Amiya Rout\.jdks\corretto-11.0.15\bin\java.exe" 
[main] INFO org.apache.kafka.clients.consumer.ConsumerConfig - ConsumerConfig values: 
    allow.auto.create.topics = true
    auto.commit.interval.ms = 5000
    auto.offset.reset = earliest
    bootstrap.servers = [127.0.0.1:9092]
    check.crcs = true
    client.dns.lookup = use_all_dns_ips
    client.id = consumer-gfg-consumer-group-1
    client.rack = 
    connections.max.idle.ms = 540000
    default.api.timeout.ms = 60000
    enable.auto.commit = true
    exclude.internal.topics = true
    fetch.max.bytes = 52428800
    fetch.max.wait.ms = 500
    fetch.min.bytes = 1
    group.id = gfg-consumer-group
    group.instance.id = null
    heartbeat.interval.ms = 3000
    interceptor.classes = []
    internal.leave.group.on.close = true
    internal.throw.on.fetch.stable.offset.unsupported = false
    isolation.level = read_uncommitted
    key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
    max.partition.fetch.bytes = 1048576
    max.poll.interval.ms = 300000
    max.poll.records = 500
    metadata.max.age.ms = 300000
    metric.reporters = []
    metrics.num.samples = 2
    metrics.recording.level = INFO
    metrics.sample.window.ms = 30000
    partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
    receive.buffer.bytes = 65536
    reconnect.backoff.max.ms = 1000
    reconnect.backoff.ms = 50
    request.timeout.ms = 30000
    retry.backoff.ms = 100
    sasl.client.callback.handler.class = null
    sasl.jaas.config = null
    sasl.kerberos.kinit.cmd = /usr/bin/kinit
    sasl.kerberos.min.time.before.relogin = 60000
    sasl.kerberos.service.name = null
    sasl.kerberos.ticket.renew.jitter = 0.05
    sasl.kerberos.ticket.renew.window.factor = 0.8
    sasl.login.callback.handler.class = null
    sasl.login.class = null
    sasl.login.refresh.buffer.seconds = 300
    sasl.login.refresh.min.period.seconds = 60
    sasl.login.refresh.window.factor = 0.8
    sasl.login.refresh.window.jitter = 0.05
    sasl.mechanism = GSSAPI
    security.protocol = PLAINTEXT
    security.providers = null
    send.buffer.bytes = 131072
    session.timeout.ms = 10000
    socket.connection.setup.timeout.max.ms = 30000
    socket.connection.setup.timeout.ms = 10000
    ssl.cipher.suites = null
    ssl.enabled.protocols = [TLSv1.2, TLSv1.3]
    ssl.endpoint.identification.algorithm = https
    ssl.engine.factory.class = null
    ssl.key.password = null
    ssl.keymanager.algorithm = SunX509
    ssl.keystore.certificate.chain = null
    ssl.keystore.key = null
    ssl.keystore.location = null
    ssl.keystore.password = null
    ssl.keystore.type = JKS
    ssl.protocol = TLSv1.3
    ssl.provider = null
    ssl.secure.random.implementation = null
    ssl.trustmanager.algorithm = PKIX
    ssl.truststore.certificates = null
    ssl.truststore.location = null
    ssl.truststore.password = null
    ssl.truststore.type = JKS
    value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version: 2.8.0
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId: ebb1d6e21cc92130
[main] INFO org.apache.kafka.common.utils.AppInfoParser - Kafka startTimeMs: 1674839122734
[main] INFO org.apache.kafka.clients.consumer.KafkaConsumer - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Subscribed to topic(s): gfg_topic
[main] INFO org.apache.kafka.clients.Metadata - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Cluster ID: orhF-HNsR465cORhmU3pTg
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Discovered group coordinator LAPTOP-FT9V6MVP:9092 (id: 2147483647 rack: null)
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] (Re-)joining group
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] (Re-)joining group
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Successfully joined group with generation Generation{generationId=1, memberId='consumer-gfg-consumer-group-1-0e33ba7e-f41f-44ff-9591-57e7ef20a164', protocol='range'}
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Finished assignment for group at generation 1: {consumer-gfg-consumer-group-1-0e33ba7e-f41f-44ff-9591-57e7ef20a164=Assignment(partitions=[gfg_topic-0])}
[main] INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Successfully synced group in generation Generation{generationId=1, memberId='consumer-gfg-consumer-group-1-0e33ba7e-f41f-44ff-9591-57e7ef20a164', protocol='range'}
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Notifying assignor about the new Assignment(partitions=[gfg_topic-0])
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Adding newly assigned partitions: gfg_topic-0
[main] INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Found no committed offset for partition gfg_topic-0
[main] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=consumer-gfg-consumer-group-1, groupId=gfg-consumer-group] Resetting offset for partition gfg_topic-0 to position FetchPosition{offset=0, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[LAPTOP-FT9V6MVP:9092 (id: 0 rack: null)], epoch=0}}.
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_0 Value: hello_neveropen 0 Partition: 0 Offset: 0
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_1 Value: hello_neveropen 1 Partition: 0 Offset: 1
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_2 Value: hello_neveropen 2 Partition: 0 Offset: 2
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_3 Value: hello_neveropen 3 Partition: 0 Offset: 3
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_4 Value: hello_neveropen 4 Partition: 0 Offset: 4
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_5 Value: hello_neveropen 5 Partition: 0 Offset: 5
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_6 Value: hello_neveropen 6 Partition: 0 Offset: 6
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_7 Value: hello_neveropen 7 Partition: 0 Offset: 7
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_8 Value: hello_neveropen 8 Partition: 0 Offset: 8
[main] INFO org.kafkademo.basics.KafkaProducerWithKeyDemo - Key: id_9 Value: hello_neveropen 9 Partition: 0 Offset: 9

Process finished with exit code 130

And you can see whatever message we have produced we can consume them using the topic name. And the same you can see in the output console. Refer to the below image.