Java consumer client in Kafka - Mini Project: Build & Apply

Java Kafka Consumer Client

📖 Scenario: You are building a simple Java application that reads messages from a Kafka topic. Kafka is a system used to send and receive messages between different parts of a program or different programs.Imagine you have a message board where people post messages, and your program needs to read those messages one by one.

🎯 Goal: Create a Java Kafka consumer client that connects to a Kafka server, subscribes to a topic named test-topic, reads messages, and prints them to the console.

📋 What You'll Learn

Create a Properties object with Kafka consumer settings

Create a KafkaConsumer<String, String> object using the properties

Subscribe the consumer to the topic test-topic

Poll messages from Kafka and print each message's key and value

Close the consumer after reading messages

💡 Why This Matters

🌍 Real World

Kafka consumers are used in real applications to process streams of data like logs, user activity, or sensor data in real time.

💼 Career

Knowing how to write Kafka consumer clients in Java is important for backend developers working with distributed systems and real-time data processing.

1

Setup Kafka consumer properties

Create a Properties object called props and set these exact properties:
bootstrap.servers to "localhost:9092",
group.id to "test-group",
key.deserializer to "org.apache.kafka.common.serialization.StringDeserializer",
value.deserializer to "org.apache.kafka.common.serialization.StringDeserializer".

Kafka

import java.util.Properties;

// Your code here

Hint

Use props.put(key, value) to set each property.

2

Create KafkaConsumer and subscribe to topic

Create a KafkaConsumer<String, String> called consumer using props. Then subscribe consumer to the topic "test-topic" using consumer.subscribe().

Kafka

import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;

// Your code here

Hint

Use Collections.singletonList("test-topic") to subscribe to one topic.

3

Poll messages and print key and value

Use consumer.poll() with a timeout of 1000 milliseconds to get records. Then use a for loop with variables record to iterate over records. Inside the loop, print the message key and value using record.key() and record.value().

Kafka

// Your code here

Hint

Use Duration.ofMillis(1000) for the poll timeout.

4

Close the consumer and print output

Call consumer.close() to close the consumer. Then print "Finished reading messages" using System.out.println().

Kafka

// Your code here

Hint

Remember to close the consumer to free resources.

Practice

1.

What is the main purpose of a Java consumer client in Kafka?

easy

A. To create Kafka topics

B. To send messages to Kafka topics

C. To read messages from Kafka topics

D. To delete Kafka topics

Solution

1. Step 1: Understand Kafka client roles

   Kafka clients include producers (send messages) and consumers (read messages).

2. Step 2: Identify the consumer's role

   The Java consumer client reads messages from Kafka topics to process or display them.

3. Final Answer:

   To read messages from Kafka topics -> Option C

4. Quick Check:

   Consumer = reads messages [OK]

Hint: Consumer reads, producer writes messages in Kafka [OK]

Common Mistakes:

• Confusing consumer with producer
• Thinking consumer creates topics
• Assuming consumer deletes topics

2.

Which of the following is the correct way to subscribe a Kafka consumer to a topic named my-topic?

consumer._____ (Collections.singletonList("my-topic"));

easy

A. subscribe

B. assign

C. poll

D. commitSync

Solution

1. Step 1: Understand consumer methods

   The subscribe method registers the consumer to listen to topics.

2. Step 2: Match method to code

   Using consumer.subscribe(Collections.singletonList("my-topic")) correctly subscribes to the topic.

3. Final Answer:

   subscribe -> Option A

4. Quick Check:

   Subscribe = listen to topics [OK]

Hint: Use subscribe() to listen to topics, not assign() [OK]

Common Mistakes:

• Using assign() instead of subscribe()
• Calling poll() to subscribe
• Using commitSync() incorrectly

3.

What will the following Java Kafka consumer code print if the topic has one message with value "Hello Kafka"?

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.value());
}

medium

A. Hello Kafka

B. null

C. ConsumerRecords object

D. No output

Solution

1. Step 1: Poll messages from topic

   The poll method fetches available messages; here it gets one message.

2. Step 2: Print each message value

   The loop prints the value() of each record, which is "Hello Kafka".

3. Final Answer:

   Hello Kafka -> Option A

4. Quick Check:

   Print record.value() = Hello Kafka [OK]

Hint: poll() fetches messages; print record.value() to see content [OK]

Common Mistakes:

• Printing the record object instead of value
• Expecting poll() to return null if no messages
• Confusing key and value fields

4.

Identify the error in this Kafka consumer code snippet:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "group1");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, String> record : records) {
    System.out.println(record.value());
}

medium

A. group.id property is missing

B. subscribe() should use assign()

C. Deserializers are incorrectly set

D. poll() argument should be Duration, not int

Solution

1. Step 1: Check poll() method signature

   In modern Kafka clients, poll requires a Duration argument, not an int.

2. Step 2: Identify incorrect argument type

   The code uses poll(100) (int), should be poll(Duration.ofMillis(100)).

3. Final Answer:

   poll() argument should be Duration, not int -> Option D

4. Quick Check:

   poll() needs Duration argument [OK]

Hint: Use Duration.ofMillis() with poll(), not plain int [OK]

Common Mistakes:

• Passing int to poll() instead of Duration
• Missing group.id property
• Using assign() instead of subscribe()

5.

You want to consume messages from two topics topicA and topicB and print their keys and values. Which code snippet correctly subscribes and processes messages?

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "group1");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer._____ (Arrays.asList("topicA", "topicB"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Key: " + record.key() + ", Value: " + record.value());
    }
}

hard

A. assign

B. subscribe

C. poll

D. commitSync

Solution

1. Step 1: Choose method to listen to multiple topics

   subscribe accepts a list of topics to listen to, while assign assigns specific partitions.

2. Step 2: Confirm correct usage in code

   Using consumer.subscribe(Arrays.asList("topicA", "topicB")) subscribes to both topics properly.

3. Final Answer:

   subscribe -> Option B

4. Quick Check:

   subscribe() listens to multiple topics [OK]

Hint: Use subscribe() with list to consume multiple topics [OK]

Common Mistakes:

• Using assign() instead of subscribe() for topics
• Calling poll() to subscribe
• Forgetting to print keys