Consumer poll loop in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer poll loop?

Consider the following Kafka consumer poll loop in Java. What will be printed to the console?

Kafka

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));
int messageCount = 0;
while (messageCount < 3) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("Received: " + record.value());
        messageCount++;
    }
}
consumer.close();

A Throws a NullPointerException because poll duration is invalid.

B Prints messages indefinitely without stopping.

C Prints exactly 3 messages from the topic and then stops.

D Prints zero messages and exits immediately.

💡 Hint

Look at the messageCount condition in the while loop.

The loop continues polling until messageCount reaches 3, so it prints exactly 3 messages and then stops.

🧠 Conceptual

intermediate

What happens if you forget to call consumer.poll() in a Kafka consumer loop?

In a Kafka consumer application, what is the effect of not calling consumer.poll() regularly inside the consumer loop?

A The consumer will automatically commit offsets without polling.

B The consumer will continue receiving messages normally without any issues.

C The consumer will throw a compile-time error.

D The consumer will stop receiving messages and may be considered dead by the broker, causing a rebalance.

💡 Hint

Think about how Kafka detects active consumers in a group.

Kafka requires consumers to poll regularly to show they are alive. If polling stops, the broker assumes the consumer is dead and triggers a rebalance.

🔧 Debug

advanced

Why does this Kafka consumer poll loop cause a commit failure?

Examine the following code snippet. Why might this consumer fail to commit offsets properly?

Kafka

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("topic1"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        process(record.value());
    }
    // Missing commit call here
}

A Offsets are never committed because commitSync or commitAsync is not called after processing.

B The consumer throws a CommitFailedException immediately.

C Offsets are committed automatically after poll, so no issue occurs.

D The consumer commits offsets before processing messages, causing data loss.

💡 Hint

Check where offset commits happen in Kafka consumer code.

Kafka consumers must explicitly commit offsets after processing messages. Without commit calls, offsets are not saved, risking duplicate processing.

📝 Syntax

advanced

Which option fixes the syntax error in this Kafka consumer poll loop?

Identify the correct fix for the syntax error in this Kafka consumer poll loop snippet:

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

A Change consumer.poll(100) to consumer.poll(Duration.ofMillis(100))

B Change consumer.poll(100) to consumer.poll(100L)

C Add a semicolon after consumer.poll(100)

D Change for loop to for (record in records)

💡 Hint

Check the expected argument type for poll method in KafkaConsumer.

The poll method expects a Duration object, not an integer. Using Duration.ofMillis(100) fixes the syntax error.

🚀 Application

expert

How to implement a graceful shutdown in a Kafka consumer poll loop?

You want to stop a Kafka consumer poll loop gracefully on receiving a shutdown signal. Which approach correctly implements this?

Kafka

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("topic"));
final AtomicBoolean running = new AtomicBoolean(true);
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    running.set(false);
    consumer.wakeup();
}));
try {
    while (running.get()) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            process(record.value());
        }
        consumer.commitSync();
    }
} catch (WakeupException e) {
    if (running.get()) throw e;
} finally {
    consumer.close();
}

A Use Thread.stop() to kill the consumer thread immediately.

B Use an AtomicBoolean flag and call consumer.wakeup() in a shutdown hook to exit poll loop safely.

C Ignore shutdown signals and let the JVM terminate the consumer abruptly.

D Call consumer.close() directly inside the poll loop without any flag or wakeup.

💡 Hint

Consider how to interrupt a blocking poll call safely.

The wakeup() method interrupts the blocking poll() call, allowing the loop to exit gracefully when the flag is set.

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