Consumer poll loop in Kafka - Commands & Configuration

Introduction

When you want to read messages from a Kafka topic, you use a consumer that keeps checking for new messages in a loop. This loop is called the consumer poll loop. It helps your app get messages as they arrive without missing any.

When you want to process new messages from a Kafka topic continuously.

When you need to build a service that reacts to events in real time.

When you want to read messages in batches to improve efficiency.

When you want to commit your read position after processing messages.

When you want to handle message processing errors gracefully.

Commands

This command starts a Kafka consumer that connects to the Kafka server at localhost:9092, subscribes to 'example-topic', and joins the consumer group 'example-group'. It polls for messages and stops after 10 seconds of inactivity.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic example-topic --group example-group --timeout-ms 10000

Expected Output

This is a sample message 1

This is a sample message 2

This is a sample message 3

→

--bootstrap-server - Specifies the Kafka server address to connect to.

→

--topic - Specifies the topic to consume messages from.

→

--group - Specifies the consumer group this consumer belongs to.

This command shows the current status of the consumer group 'example-group', including which partitions it is reading and the offset positions. It helps verify that the consumer poll loop is working and tracking progress.

Terminal

kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group example-group

Expected Output

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID                                    HOST            CLIENT-ID

example-group   example-topic   0          15              15              0    consumer-1-12345-67890abcdef                  /127.0.0.1      consumer-1

→

--describe - Shows detailed information about the consumer group.

→

--group - Specifies the consumer group to describe.

Key Concept

If you remember nothing else from this pattern, remember: the consumer poll loop continuously checks for new messages so your app can process them in real time without missing any.

Common Mistakes

Not running the consumer in a loop and polling repeatedly.

Without a loop, the consumer reads messages only once and then stops, missing future messages.

Use a loop that calls poll() repeatedly to keep reading new messages as they arrive.

Not committing offsets after processing messages.

If offsets are not committed, the consumer may re-read the same messages after restart, causing duplicate processing.

Commit offsets manually or enable auto-commit to track progress and avoid duplicates.

Setting poll timeout too low or too high without testing.

Too low timeout causes frequent empty polls wasting resources; too high causes delays in processing new messages.

Choose a balanced poll timeout based on message frequency and processing speed.

Summary

Use the kafka-console-consumer command with --topic and --group to start a consumer that polls messages in a loop.

Check consumer group status with kafka-consumer-groups to verify message consumption and offset progress.

Remember to poll repeatedly and commit offsets to avoid missing or duplicating messages.

Practice

1. What is the main purpose of using a poll() method inside a Kafka consumer loop?

easy

A. To send messages to Kafka topics

B. To continuously fetch new messages from Kafka topics

C. To close the consumer connection

D. To create a new Kafka topic

Solution

1. Step 1: Understand the role of poll() in Kafka consumer

   The poll() method is used to fetch messages from Kafka topics.

2. Step 2: Recognize the loop usage

   Using poll() inside a loop allows the consumer to keep listening and fetching new messages continuously.

3. Final Answer:

   To continuously fetch new messages from Kafka topics -> Option B

4. Quick Check:

   poll() fetches messages repeatedly [OK]

Hint: poll() inside loop fetches messages repeatedly [OK]

Common Mistakes:

• Thinking poll() sends messages
• Confusing poll() with closing connection
• Assuming poll() creates topics

2. Which of the following is the correct syntax to start a Kafka consumer poll loop in Java?

easy

A. consumer.poll(Duration(100));

B. for(int i=0; i<10; i++) consumer.poll(100);

C. consumer.poll(100); while(true) {}

D. while(true) { consumer.poll(Duration.ofMillis(100)); }

Solution

1. Step 1: Check correct poll method usage

   The poll method requires a Duration argument like Duration.ofMillis(100) in Java.

2. Step 2: Verify loop structure

   A continuous loop like while(true) is used to keep polling messages repeatedly.

3. Final Answer:

   while(true) { consumer.poll(Duration.ofMillis(100)); } -> Option D

4. Quick Check:

   Correct poll syntax with continuous loop [OK]

Hint: Use while(true) with poll(Duration.ofMillis(x)) [OK]

Common Mistakes:

• Using poll with int instead of Duration
• Not using a loop for continuous polling
• Incorrect Duration constructor syntax

3. What will be the output of the following Kafka consumer poll loop snippet?

while (true) {
  var records = consumer.poll(Duration.ofMillis(100));
  System.out.println(records.count());
  break;
}

medium

A. Prints the number of records fetched in one poll and exits

B. Prints zero and continues infinitely

C. Throws an exception because of missing commit

D. Does not compile due to syntax error

Solution

1. Step 1: Analyze the poll call and record count

   The poll() fetches records and records.count() returns how many were fetched.

2. Step 2: Understand the loop and break

   The loop runs once because of break; after printing count, so it prints the count once and stops.

3. Final Answer:

   Prints the number of records fetched in one poll and exits -> Option A

4. Quick Check:

   poll() fetches records, break stops loop [OK]

Hint: Break after poll prints count once [OK]

Common Mistakes:

• Assuming infinite loop without break
• Thinking commit is mandatory to print count
• Confusing syntax causing compile error

4. Identify the error in this Kafka consumer poll loop code snippet:

while (true) {
  var records = consumer.poll(100);
  for (var record : records) {
    System.out.println(record.value());
  }
}

medium

A. poll() argument should be a Duration, not an int

B. for-each loop syntax is incorrect

C. Missing consumer.subscribe() call before loop

D. System.out.println cannot print record values

Solution

1. Step 1: Check poll method argument type

   The poll() method requires a Duration argument, not a plain integer.

2. Step 2: Confirm other code parts

   The for-each loop and print statement are correct; missing subscribe is not shown as error here.

3. Final Answer:

   poll() argument should be a Duration, not an int -> Option A

4. Quick Check:

   poll() needs Duration argument [OK]

Hint: poll() needs Duration, not int [OK]

Common Mistakes:

• Passing int instead of Duration to poll()
• Confusing for-each syntax
• Assuming subscribe() missing causes compile error

5. You want to process messages from multiple Kafka topics continuously and commit offsets manually after processing. Which code snippet correctly implements this using a consumer poll loop?

hard

A. consumer.poll(Duration.ofMillis(200));
consumer.commitSync();

B. while(true) {
var records = consumer.poll(200);
for(var record : records) {
process(record);
}
consumer.commitAsync();
}

C. while(true) {
var records = consumer.poll(Duration.ofMillis(200));
for(var record : records) {
process(record);
}
consumer.commitSync();
}

D. for(int i=0; i<10; i++) {
var records = consumer.poll(Duration.ofMillis(200));
consumer.commitSync();
}

Solution

1. Step 1: Check poll argument and loop structure

   while(true) { var records = consumer.poll(Duration.ofMillis(200)); for(var record : records) { process(record); } consumer.commitSync(); } uses poll(Duration.ofMillis(200)) inside a continuous while(true) loop, which is correct.

2. Step 2: Verify message processing and manual commit

   It processes each record and calls commitSync() after processing, ensuring offsets are committed manually.

3. Final Answer:

   while(true) { var records = consumer.poll(Duration.ofMillis(200)); for(var record : records) { process(record); } consumer.commitSync(); } -> Option C

4. Quick Check:

   Continuous poll, process, then manual commit [OK]

Hint: Use while(true), poll with Duration, then commitSync() [OK]

Common Mistakes:

• Passing int instead of Duration to poll()
• Using for loop instead of continuous loop
• Not processing records before commit
• Missing loop causing single poll only