Consumer poll loop in Kafka - Mini Project: Build & Apply

Consumer poll loop

📖 Scenario: You are building a simple Kafka consumer application that reads messages from a Kafka topic continuously. This is common in real-world systems where you want to process data streams as they arrive.

🎯 Goal: Build a Kafka consumer poll loop that continuously polls messages from a topic called test-topic and prints each message's key and value.

📋 What You'll Learn

Create a Kafka consumer with the group ID my-group.

Subscribe the consumer to the topic test-topic.

Set a poll timeout of 100 milliseconds.

Use a while True loop to continuously poll messages.

Print each message's key and value in the format: Key: {key}, Value: {value}.

💡 Why This Matters

🌍 Real World

Kafka consumers are used in real-time data processing systems to read streams of data such as logs, user activity, or sensor data.

💼 Career

Understanding how to write a consumer poll loop is essential for backend developers, data engineers, and anyone working with streaming data platforms.

1

Create Kafka consumer and subscribe to topic

Create a Kafka consumer called consumer with the group ID my-group and bootstrap server localhost:9092. Then subscribe the consumer to the topic test-topic.

Kafka

# Create Kafka consumer with group ID 'my-group' and subscribe to 'test-topic'
# Your code here

Hint

Use KafkaConsumer from the kafka library. You can pass the topic name directly or use consumer.subscribe() after creating the consumer.

2

Set poll timeout variable

Create a variable called poll_timeout_ms and set it to 100 to represent the poll timeout in milliseconds.

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

# Set poll timeout to 100 milliseconds
# Your code here

Hint

This variable will be used as the timeout value when polling messages.

3

Write the consumer poll loop

Write a while True loop that polls messages from consumer using poll_timeout_ms as the timeout. Iterate over the records returned by consumer.poll(timeout_ms=poll_timeout_ms) using nested loops (for partition_records in records.values(): for message in partition_records:) and print each message's key and value using print(f"Key: {message.key}, Value: {message.value}").

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

poll_timeout_ms = 100

# Write a while True loop to poll messages and print key and value
# Your code here

Hint

Use nested loops because consumer.poll() returns a dictionary of partitions to lists of messages.

4

Print the consumer output

Run the program and observe the printed output lines showing message keys and values in the format Key: {key}, Value: {value}. For example, if a message has key b'user1' and value b'hello', the output should be Key: b'user1', Value: b'hello'.

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

poll_timeout_ms = 100

while True:
    records = consumer.poll(timeout_ms=poll_timeout_ms)
    for partition_records in records.values():
        for message in partition_records:
            print(f"Key: {message.key}, Value: {message.value}")

# Run the program and observe output

Hint

Make sure your Kafka broker is running and the topic test-topic has messages with keys and values.

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