Consumer poll loop in Kafka - Step-by-Step Execution

Process Flow - Consumer poll loop

Start Consumer

↓

Poll for messages

↓

Are messages received?

No→Wait and Poll again

Yes↓

Process each message

↓

Commit offsets

↓

Loop back to Poll

The consumer starts, polls messages in a loop, processes them if any, commits offsets, then polls again.

Execution Sample

Kafka

consumer = KafkaConsumer(bootstrap_servers=['localhost:9092'], enable_auto_commit=False)
consumer.subscribe(['topic'])
while True:
    records = consumer.poll(timeout_ms=1000)
    for record_list in records.values():
        for record in record_list:
            process(record)
    consumer.commit()

This code continuously polls messages from a Kafka topic, processes them, and commits offsets.

Process Table

Step	Action	Poll Result	Process Messages	Commit Offsets	Loop Back
1	Start consumer and poll	No messages	No processing	No commit	Poll again
2	Poll messages	Received 2 messages	Process message 1	No commit yet	Continue processing
3	Process message 2	Same 2 messages	Process message 2	No commit yet	After processing all
4	Commit offsets	No new poll	No processing	Offsets committed	Poll again
5	Poll messages	No messages	No processing	No commit	Poll again
6	Poll messages	Received 1 message	Process message 1	No commit yet	Continue processing
7	Commit offsets	No new poll	No processing	Offsets committed	Poll again
8	Poll messages	No messages	No processing	No commit	Poll again
Exit	Stop condition met (not shown in code)	Stop polling	Stop processing	Stop committing	Exit loop

💡 Loop runs indefinitely unless stopped externally; here exit is hypothetical.

Status Tracker

Variable	Start	After 1	After 2	After 3	After 4	After 5	After 6	Final
records	empty	empty	2 messages	2 messages	empty	empty	1 message	empty
processed_messages	0	0	1	2	2	2	1	1
offsets_committed	false	false	false	false	true	true	true	true

Key Moments - 3 Insights

Why does the consumer poll again even if no messages are received?

Because the poll loop is continuous; when poll returns no messages (see Step 1 and 5), the loop waits and polls again to check for new messages.

When are offsets committed in the loop?

Offsets are committed only after all messages from the current poll are processed (see Step 4 and 7), ensuring processed messages are marked.

Does the poll method return messages immediately every time?

No, poll waits up to the timeout and may return no messages if none are available (see Step 1 and 5).

Visual Quiz - 3 Questions

Look at the execution table, what is the value of 'processed_messages' after Step 3?

A 2

B 1

C 0

D 3

💡 Hint

Check the variable_tracker row for 'processed_messages' at After 3.

At which step does the consumer commit offsets for the first time?

A Step 2

B Step 4

C Step 3

D Step 1

💡 Hint

Look at the 'Commit Offsets' column in the execution_table.

If the poll always returned no messages, how would the 'processed_messages' variable change?

A It would keep increasing

B It would reset to 0 each loop

C It would stay at 0

D It would become negative

💡 Hint

Refer to variable_tracker and execution_table rows where poll returns no messages.

Concept Snapshot

Consumer poll loop syntax:
while True:
  records = consumer.poll(timeout_ms=1000)
  for record_list in records.values():
    for record in record_list:
      process(record)
  consumer.commit()

Behavior:
- Poll waits for messages
- Processes all received messages
- Commits offsets after processing
- Loops continuously to consume new messages

Full Transcript

This visual execution shows how a Kafka consumer uses a poll loop to receive messages. The consumer starts and polls for messages. If no messages arrive, it polls again. When messages arrive, it processes each one, then commits offsets to mark progress. This repeats indefinitely. Variables like 'records', 'processed_messages', and 'offsets_committed' change step by step. Key moments clarify why polling repeats without messages and when commits happen. The quiz tests understanding of these steps and variable changes.

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