Consumer poll loop in Kafka - Time & Space Complexity
When using Kafka, the consumer poll loop is how your program reads messages continuously.
We want to know how the time to process messages grows as more messages arrive.
Analyze the time complexity of the following code snippet.
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
process(record);
}
}
This code keeps polling Kafka for new messages and processes each message one by one.
Identify the loops, recursion, array traversals that repeat.
- Primary operation: The for-loop that processes each message in the batch.
- How many times: Once for every message received in each poll call.
Explain the growth pattern intuitively.
| Input Size (n) | Approx. Operations |
|---|---|
| 10 messages | Processes 10 messages |
| 100 messages | Processes 100 messages |
| 1000 messages | Processes 1000 messages |
Pattern observation: The time grows directly with the number of messages to process.
Time Complexity: O(n)
This means the time to process messages grows linearly with how many messages arrive.
[X] Wrong: "The poll loop always takes the same time no matter how many messages come in."
[OK] Correct: Actually, the more messages received in each poll, the longer the processing takes because each message is handled one by one.
Understanding how your consumer handles messages helps you explain how your program scales with load, a key skill in real projects.
"What if we processed messages in parallel inside the poll loop? How would the time complexity change?"
Practice
poll() method inside a Kafka consumer loop?Solution
Step 1: Understand the role of
Thepoll()in Kafka consumerpoll()method is used to fetch messages from Kafka topics.Step 2: Recognize the loop usage
Usingpoll()inside a loop allows the consumer to keep listening and fetching new messages continuously.Final Answer:
To continuously fetch new messages from Kafka topics -> Option BQuick Check:
poll() fetches messages repeatedly [OK]
- Thinking poll() sends messages
- Confusing poll() with closing connection
- Assuming poll() creates topics
Solution
Step 1: Check correct poll method usage
The poll method requires a Duration argument likeDuration.ofMillis(100)in Java.Step 2: Verify loop structure
A continuous loop likewhile(true)is used to keep polling messages repeatedly.Final Answer:
while(true) { consumer.poll(Duration.ofMillis(100)); } -> Option DQuick Check:
Correct poll syntax with continuous loop [OK]
- Using poll with int instead of Duration
- Not using a loop for continuous polling
- Incorrect Duration constructor syntax
while (true) {
var records = consumer.poll(Duration.ofMillis(100));
System.out.println(records.count());
break;
}Solution
Step 1: Analyze the poll call and record count
Thepoll()fetches records andrecords.count()returns how many were fetched.Step 2: Understand the loop and break
The loop runs once because ofbreak;after printing count, so it prints the count once and stops.Final Answer:
Prints the number of records fetched in one poll and exits -> Option AQuick Check:
poll() fetches records, break stops loop [OK]
- Assuming infinite loop without break
- Thinking commit is mandatory to print count
- Confusing syntax causing compile error
while (true) {
var records = consumer.poll(100);
for (var record : records) {
System.out.println(record.value());
}
}Solution
Step 1: Check poll method argument type
Thepoll()method requires aDurationargument, not a plain integer.Step 2: Confirm other code parts
The for-each loop and print statement are correct; missing subscribe is not shown as error here.Final Answer:
poll() argument should be a Duration, not an int -> Option AQuick Check:
poll() needs Duration argument [OK]
- Passing int instead of Duration to poll()
- Confusing for-each syntax
- Assuming subscribe() missing causes compile error
Solution
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(); } usespoll(Duration.ofMillis(200))inside a continuouswhile(true)loop, which is correct.Step 2: Verify message processing and manual commit
It processes each record and callscommitSync()after processing, ensuring offsets are committed manually.Final Answer:
while(true) { var records = consumer.poll(Duration.ofMillis(200)); for(var record : records) { process(record); } consumer.commitSync(); } -> Option CQuick Check:
Continuous poll, process, then manual commit [OK]
- 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
Ready to prove it?
Pass the test and this pattern moves unknown
