Java consumer client in Kafka - Step-by-Step Execution

Process Flow - Java consumer client

Start Java Consumer

↓

Configure Properties

↓

Create KafkaConsumer

↓

Subscribe to Topic

↓

Poll for Records

↓

Process Records

↓

Commit Offsets

↓

Repeat Poll or Exit

↓

Close Consumer

This flow shows how a Java Kafka consumer starts, subscribes to a topic, polls messages, processes them, commits offsets, and closes.

Execution Sample

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
consumer.subscribe(Collections.singletonList("my-topic"));
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

This code sets up a Kafka consumer, subscribes to 'my-topic', and polls for messages.

Process Table

Step	Action	Evaluation	Result
1	Create Properties object	Properties empty	Properties with bootstrap.servers and group.id set
2	Create KafkaConsumer with properties and deserializers	KafkaConsumer instance created	Consumer ready to subscribe
3	Subscribe to topic 'my-topic'	Subscription successful	Consumer listens to 'my-topic'
4	Poll for records with 100ms timeout	Poll sent to Kafka broker	ConsumerRecords received (may be empty)
5	Process each record	Iterate over records	Records processed (e.g., print key/value)
6	Commit offsets	Offsets committed synchronously	Kafka knows messages processed
7	Repeat poll or exit	Loop continues or stops	Consumer continues polling or closes
8	Close consumer	Consumer closed	Resources freed, connection closed

💡 Consumer closes after processing or on application shutdown

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	After Step 5	After Step 6	Final
props	empty	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	{"bootstrap.servers":"localhost:9092","group.id":"test-group"}	unchanged
consumer	null	null	KafkaConsumer instance	KafkaConsumer subscribed	KafkaConsumer polled	KafkaConsumer processed records	KafkaConsumer committed offsets	KafkaConsumer closed
records	null	null	null	ConsumerRecords (may be empty or contain messages)	Processed records	Processed records	Processed records	null

Key Moments - 3 Insights

Why do we need to set 'group.id' in the properties?

The 'group.id' identifies the consumer group this client belongs to, which Kafka uses to manage message delivery and offset commits. See execution_table step 1 where properties are set.

What happens if the poll() call returns no records?

If poll() returns no records, the consumer waits and polls again in the loop. This is normal if no new messages are available. See execution_table step 4.

Why must we commit offsets after processing records?

Committing offsets tells Kafka which messages have been processed, so on restart the consumer doesn't reprocess them. See execution_table step 6.

Visual Quiz - 3 Questions

Look at the execution_table at step 4, what does the poll() method return?

A A ConsumerRecords object containing messages or empty

B A Properties object with configuration

C A KafkaConsumer instance

D A list of topic names

💡 Hint

Refer to execution_table row 4 under Result column

At which step is the consumer subscribed to the topic?

A Step 2

B Step 3

C Step 5

D Step 6

💡 Hint

Check execution_table row 3 Action and Result columns

If we omit committing offsets, what will likely happen on consumer restart?

A Consumer will skip messages already processed

B Consumer will reprocess messages from last committed offset

C Consumer will fail to connect to Kafka

D Consumer will automatically commit offsets

💡 Hint

See key_moments answer about offset commits and execution_table step 6

Concept Snapshot

Java Kafka Consumer Client:
- Configure Properties (bootstrap.servers, group.id)
- Create KafkaConsumer with deserializers
- Subscribe to topic(s)
- Poll for messages in a loop
- Process messages
- Commit offsets to track progress
- Close consumer on shutdown

Full Transcript

This visual execution trace shows how a Java Kafka consumer client works step-by-step. First, properties are configured with the Kafka server address and consumer group ID. Then, a KafkaConsumer instance is created using these properties and deserializers for keys and values. The consumer subscribes to a topic, for example 'my-topic'. It then polls the Kafka broker for new messages with a timeout. If messages are received, the consumer processes each record, such as printing the key and value. After processing, the consumer commits the offsets to Kafka to mark messages as consumed. This polling and processing loop continues until the consumer is closed, freeing resources and closing connections. Key points include the importance of group.id for consumer groups, handling empty polls, and committing offsets to avoid reprocessing messages on restart.

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