In this post we are going to cover consumers and consumer groups: how a group assigns partitions, offset commits, auto.offset.reset, rebalances, consumer lag, and the new rebalance protocol and share groups. These are 7 real Kafka interview questions for data engineers, asked the way they come up on a real project, and the last one is a scenario where a consumer writes to a database and must not lose or double-count records when it crashes.
A consumer reads records from Kafka, and a consumer group is a set of consumers that share the work of reading a topic. Most production problems on the read side, like idle consumers, reprocessed messages, missed data and rebalance storms, come from how groups and offsets work. That is why interviewers spend a lot of time here.
Part 3 of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT.
- The short answer
- Let's set up the cluster and some sample data
- Question 1: You add a second consumer and it gets no messages. How do consumer groups assign partitions?
- Question 2: After a crash, the consumer reprocessed a thousand messages. How do offset commits work?
- Question 3: A new consumer group started reading from the end and missed old data. What does auto offset reset do?
- Question 4: The consumer group keeps rebalancing and processing stalls. What causes rebalances, and how do you reduce them?
- Question 5: Your manager asks how far behind the consumers are. How do you measure and act on consumer lag?
- Question 6: A teammate heard about the new consumer rebalance protocol and share groups. What are they, and when would you use them?
- Scenario 7: A consumer that writes to a database must not lose or double-count records when it crashes. How do you design it?
- The full script
The short answer
A consumer group gives each partition to exactly one member and resumes from the committed offset, so commit after processing and expect at-least-once. auto.offset.reset applies only without a valid committed offset. Rebalances come from joins, leaves, timeouts and slow polls; cooperative-sticky or KIP-848 make them incremental, and share groups let more consumers than partitions share independent work.
Let's set up the cluster and some sample data
We run a local Apache Kafka 4.4 cluster in KRaft mode with 1 node, which is both the broker and the controller. Here are the sample files: small text files of invoices, clicks, tasks and payments, and one short Python script per question.
invoice-1
invoice-2
invoice-3
invoice-4
invoice-5
invoice-6
invoice-7
invoice-8
invoice-9
invoice-10from confluent_kafka import Consumer
def member(name):
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'invoicing', 'client.id': name})
c.subscribe(['invoices'])
return c
a, b = member('consumer-a'), member('consumer-b')
for _ in range(10): # poll both while the group forms
a.poll(0.5)
b.poll(0.5)
for name, c in [('consumer-a', a), ('consumer-b', b)]:
parts = sorted(tp.partition for tp in c.assignment())
print(name, 'partitions', parts)
a.close()
b.close()import os, sys
from confluent_kafka import Consumer
crash_at = int(sys.argv[1]) if len(sys.argv) > 1 else -1
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'billing', 'enable.auto.commit': False,
'auto.offset.reset': 'earliest',
'session.timeout.ms': 6000}) # lab: fast detection
c.subscribe(['bills'])
done, idle = [], 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1 if c.assignment() else 0
continue
done.append(m.offset()) # process the bill
if m.offset() == crash_at:
print('processed', done, '-> crash')
os._exit(1)
if len(done) % 4 == 0: # commit every 4 records
c.commit(message=m, asynchronous=False)
if done:
c.commit(asynchronous=False) # the rest, before leaving
print('processed', done)
c.close()click-1
click-2
click-3
click-4
click-5import sys
from confluent_kafka import Consumer
group, reset = sys.argv[1], sys.argv[2]
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': group, 'auto.offset.reset': reset})
c.subscribe(['clicks'])
got, idle = [], 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
got.append(m.value().decode())
c.close()
print(f'group {group}, reset {reset}: read {len(got)}', got)import sys
from confluent_kafka import Consumer
mode = sys.argv[1] # range | cooperative-sticky | consumer
if mode == 'consumer':
extra = {'group.protocol': 'consumer'} # KIP-848
else:
extra = {'partition.assignment.strategy': mode}
log = {'consumer-a': [], 'consumer-b': []}
def member(name):
def note(what):
def callback(consumer, parts):
ids = sorted(p.partition for p in parts)
if ids:
log[name].append((what, ids))
return callback
c = Consumer({'bootstrap.servers': 'localhost:9092',
'client.id': name, 'group.id': 'rebalance-' + mode,
**extra})
c.subscribe(['orders'], on_assign=note('assigned'),
on_revoke=note('revoked'))
return c
a = member('consumer-a')
for _ in range(8):
a.poll(0.5)
b = member('consumer-b') # a second consumer joins
for _ in range(16):
a.poll(0.5)
b.poll(0.5)
for name, events in log.items():
print(mode, name, events)
a.close()
b.close()import time
from confluent_kafka import Consumer
def show(what):
return lambda c, parts: print(what, [p.partition for p in parts])
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'slow', 'auto.offset.reset': 'earliest',
'enable.auto.commit': False,
'session.timeout.ms': 6000, # lab: default 45 s
'max.poll.interval.ms': 6000}) # lab: default 5 min
c.subscribe(['bills'], on_assign=show('assigned'),
on_revoke=show('revoked'), on_lost=show('lost'))
seen = 0
while seen < 2:
m = c.poll(1.0)
if m is None:
continue
if m.error():
print('poll error:', m.error().name())
continue
seen += 1
print('processing offset', m.offset())
if seen == 1:
time.sleep(8) # slow processing
c.close()import sys
from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
first, count = int(sys.argv[1]), int(sys.argv[2])
for i in range(first, first + count):
p.produce('events', value=f'event-{i}', partition=i % 3)
p.flush()
print('produced', count, 'events')from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'reporting', 'auto.offset.reset': 'earliest'})
c.subscribe(['events'])
n, idle = 0, 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
n += 1
c.close() # commits the final offsets
print('reporting read', n, 'events')from confluent_kafka import Consumer, TopicPartition
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'reporting'})
parts = [TopicPartition('events', p) for p in range(3)]
total = 0
for tp in c.committed(parts, timeout=10):
low, high = c.get_watermark_offsets(tp, timeout=10)
lag = high - tp.offset
total += lag
print(f'partition {tp.partition}: committed {tp.offset}, '
f'end {high}, lag {lag}')
print('total lag', total)
c.close()from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'new-protocol',
'group.protocol': 'consumer', # opt in to KIP-848
'auto.offset.reset': 'earliest'})
c.subscribe(['events'])
n, idle = 0, 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
n += 1
c.close()
print('new-protocol read', n, 'events')task-1
task-2
task-3
task-4
task-5
task-6pay-1,100
pay-2,250
pay-3,75
pay-4,40
pay-5,310
pay-6,60create table if not exists naive (payment_id text, amount int);
create table if not exists safe (
payment_id text primary key, amount int);import os, sqlite3, sys
from confluent_kafka import Consumer
UPSERT = ('insert into safe values (?, ?) on conflict (payment_id) '
'do update set amount = excluded.amount')
db = sqlite3.connect('shop.db')
db.executescript(open('tables.sql').read())
crash_at = int(sys.argv[1]) if len(sys.argv) > 1 else -1
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'db-sink', 'enable.auto.commit': False,
'auto.offset.reset': 'earliest',
'session.timeout.ms': 6000}) # lab: fast detection
c.subscribe(['payments'])
while msgs := c.consume(10, timeout=10):
for m in msgs:
if m.error():
raise SystemExit(m.error())
row = m.value().decode().split(',') # payment_id, amount
with db: # one DB transaction
db.execute('insert into naive values (?, ?)', row)
db.execute(UPSERT, row)
if m.offset() == crash_at:
print('crash after the DB commit of offset', m.offset())
os._exit(1)
if m.offset() % 2 == 1: # commit Kafka every 2
c.commit(message=m, asynchronous=False)
c.close()import sqlite3
db = sqlite3.connect('shop.db')
for table in ['naive', 'safe']:
q = f'select count(*), sum(amount) from {table}'
print(table, 'rows %d, total amount %d' % db.execute(q).fetchone())Question 1: You add a second consumer and it gets no messages. How do consumer groups assign partitions?
In a consumer group, each partition is assigned to exactly one member. So a group can't use more consumers than the topic has partitions.
Let's look at the usual cause. We create invoices without a partition count, describe it, and load ten invoices.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic invoices
Created topic invoices.
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic invoices
Topic: invoices TopicId: <topic-id> PartitionCount: 1 ReplicationFactor: 1 Configs:
min.insync.replicas=1
Topic: invoices Partition: 0 Leader: 1 Replicas: 1 Isr: 1 Elr: LastKnownElr:
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic invoices < invoices.txtIt got the broker default: PartitionCount: 1. Auto-created topics get the same default. Now two consumers join the group invoicing with python two.py.
$ python two.py
consumer-a partitions [0]
consumer-b partitions []Consumer A gets partition 0, and consumer B gets nothing. It is an idle standby: it helps on failover, but adds no throughput. Let's add a second partition and start the group again.
$ kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic invoices --partitions 2
$ python two.py
consumer-a partitions [0]
consumer-b partitions [1]Now A gets partition 0 and B gets partition 1. The default range assignor sorts the members and splits the partitions in ranges.
Size partitions for the most consumers you will need, and remember that adding partitions moves keys. If you need more workers than partitions, and no ordering, look at share groups in Question 6.
Question 2: After a crash, the consumer reprocessed a thousand messages. How do offset commits work?
A consumer group commits, per partition, the offset of the next record to read. Kafka stores it in the internal __consumer_offsets topic, and after a restart or a rebalance, the group resumes from there. So everything processed after the last commit runs again.
Let's see it. First we create a bills topic with ten records.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic bills --partitions 1
Created topic bills.
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic bills < invoices.txtOur worker commits every four records, and we tell it to crash at offset 6 with python worker.py 6.
$ python worker.py 6
processed [0, 1, 2, 3, 4, 5, 6] -> crashIt processed offsets 0 to 6, then crashed. Now we describe the group and look for the offsets topic.
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID
HOST CLIENT-ID
billing bills 0 4 10 6
rdkafka-<uuid> /127.0.0.1 rdkafka
$ kafka-topics.sh --bootstrap-server localhost:9092 --list | grep consumer_offsets
__consumer_offsetsThe committed offset is 4, with a lag of 6. Now we restart the worker.
$ python worker.py
processed [4, 5, 6, 7, 8, 9]
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing
Consumer group 'billing' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID
HOST CLIENT-ID
billing bills 0 10 10 0 -
- -It processes 4, 5 and 6 again, then the rest, and the lag is 0. That is at-least-once: nothing lost, but duplicates.
With auto commit, the default, offsets are committed every five seconds, so a crash replays up to five seconds of records. At a high rate, that is your thousand messages.
Commit after processing, in batches you can afford to redo, and make the processing idempotent. In librdkafka, auto commit can even commit records you have not finished, so set enable.auto.offset.store=false and call store_offsets() after processing.
Question 3: A new consumer group started reading from the end and missed old data. What does auto offset reset do?
auto.offset.reset decides where a group starts when it has no committed offset, or when its offset is out of range. The default is latest, in the Java client and in librdkafka, so a new group starts at the end.
Let's try it. We load five clicks, then read with a new group dashboard on latest and a new group backfill on earliest.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic clicks --partitions 1
Created topic clicks.
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic clicks < clicks.txt
$ python reader.py dashboard latest
group dashboard, reset latest: read 0 []
$ python reader.py backfill earliest
group backfill, reset earliest: read 5 ['click-1', 'click-2', 'click-3', 'click-4', 'click-5']dashboard reads 0 and backfill reads all 5. Since Kafka 4.0, the Java client also accepts by_duration. Let's start a Java console consumer from one hour ago.
$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic clicks --group last-hour --command-property auto.offset.reset=by_duration:PT1H --max-messages 5
The consumer rebalance protocol (KIP-848) is production-ready! Set group.protocol=consumer to try it
out. See https://kafka.apache.org/documentation/#consumer_rebalance_protocol
click-1
click-2
click-3
click-4
click-5It reads all five clicks. The docs recommend by_duration over latest for topics whose partition count can grow, and it needs synced clocks.
Once a group has committed, the reset setting doesn't matter. We add click-6 and restart backfill with latest.
$ echo click-6 | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic clicks
$ python reader.py backfill latest
group backfill, reset latest: read 1 ['click-6']It reads only click-6, from its committed offset. To fix a group that already skipped history, we reset its offsets. Let's move dashboard to earliest.
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group dashboard --topic clicks --to-earliest --execute
GROUP TOPIC PARTITION NEW-OFFSET
dashboard clicks 0 0
$ python reader.py dashboard latest
group dashboard, reset latest: read 6 ['click-1', 'click-2', 'click-3', 'click-4', 'click-5', 'click-6']The new offset is 0, and dashboard reads all 6 clicks.
Set auto.offset.reset explicitly on every consumer.
Question 4: The consumer group keeps rebalancing and processing stalls. What causes rebalances, and how do you reduce them?
A rebalance reassigns partitions. It happens when:
- A member joins or leaves
- A member misses
session.timeout.ms- 45 s default - Longer than
max.poll.interval.msbetween polls - 5 min default - Partitions or subscriptions change
How much it hurts depends on the protocol. Let's create a topic with three partitions and let a second consumer join, once with eager range, once with cooperative-sticky, and once with the new consumer protocol.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3
Created topic orders.
$ python rebalance.py range
range consumer-a [('assigned', [0, 1, 2]), ('revoked', [0, 1, 2]), ('assigned', [0, 1])]
range consumer-b [('assigned', [2])]
$ python rebalance.py cooperative-sticky
cooperative-sticky consumer-a [('assigned', [0, 1, 2]), ('revoked', [0])]
cooperative-sticky consumer-b [('assigned', [0])]
$ python rebalance.py consumer
consumer consumer-a [('assigned', [0, 1, 2]), ('revoked', [2])]
consumer consumer-b [('assigned', [2])]With range, consumer A gives up all three partitions and then gets two back. Everything stops in between. Cooperative-sticky moves only what has to move: A gives up just partition 0. The new consumer protocol, KIP-848, is incremental too, and the broker computes the assignment: here, A gives up partition 2.
The classic stall is slow processing. slow.py has a max.poll.interval.ms of six seconds, a lab value, and spends eight seconds on one record.
$ python slow.py 2> slow.log
assigned [0]
processing offset 0
poll error: _MAX_POLL_EXCEEDED
lost [0]
assigned [0]
processing offset 0
revoked [0]We get _MAX_POLL_EXCEEDED. The consumer is removed from the group, loses its partition, rejoins, and gets offset 0 again.
To reduce rebalances:
- Short work per poll: smaller
max.poll.records, orpause()/resume() - Static membership:
group.instance.idfor rolling restarts - cooperative-sticky, or
group.protocol=consumer(KIP-848)
Question 5: Your manager asks how far behind the consumers are. How do you measure and act on consumer lag?
Lag is the log end offset minus the group's committed offset, per partition. It tells us how many records are waiting.
First, nine events over three partitions, and the reporting group reads them all.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic events --partitions 3
Created topic events.
$ python produce.py 0 9
produced 9 events
$ python catchup.py
reporting read 9 eventsIt read 9 events, so it has caught up. Then reporting goes down while twelve more events arrive, and we describe the group.
$ python produce.py 9 12
produced 12 events
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group reporting
Consumer group 'reporting' has no active members.
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID
HOST CLIENT-ID
reporting events 0 3 7 4 -
- -
reporting events 1 3 7 4 -
- -
reporting events 2 3 7 4 -
- -Each partition shows CURRENT-OFFSET 3, LOG-END-OFFSET 7 and LAG 4. Let's get the same numbers from code with python lag.py, using committed offsets and watermarks.
$ python lag.py
partition 0: committed 3, end 7, lag 4
partition 1: committed 3, end 7, lag 4
partition 2: committed 3, end 7, lag 4
total lag 12Lag 4 per partition, total lag 12. Consumers also expose a records-lag-max metric, and lag exporters read these offsets for dashboards.
Don't alert on one number. Alert when lag keeps growing, or when the oldest waiting record gets too old for the business.
To act on it, we add consumers up to the partition count, make processing faster, batch the writes to the sink, or add partitions if we are at the limit.
Check for one hot partition. More consumers won't fix that.
Question 6: A teammate heard about the new consumer rebalance protocol and share groups. What are they, and when would you use them?
The new consumer rebalance protocol, KIP-848, is GA since Kafka 4.0. The broker computes the assignment and rebalances member by member, with no global stop. Share groups, KIP-932, are queues for Kafka, production-ready since 4.2.
| Classic group | Consumer protocol (KIP-848) | Share group (KIP-932) | |
|---|---|---|---|
| Status | the default | GA since 4.0, opt in | production-ready in 4.2 |
| Assignment | client-side assignor | on the broker: uniform, range | on the broker, partitions shared |
| Active consumers | up to the partitions | up to the partitions | can exceed the partitions |
| Ordering | per partition | per partition | no per-key order |
You opt in with group.protocol=consumer. Let's run a consumer that does, and look at the group state.
$ python kip848.py
new-protocol read 21 events
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group new-protocol --state
Consumer group 'new-protocol' has no active members.
GROUP COORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS
new-protocol localhost:9092 (1) uniform Empty 0It read 21 events, and the state shows the server-side assignor, uniform. The Java default is still classic, with the switch planned for 5.0. That is the banner the Java console consumer printed in Question 3.
With the new protocol, session.timeout.ms, heartbeat.interval.ms and partition.assignment.strategy no longer apply. The broker has group.consumer.* settings, and group.remote.assignor picks a server-side assignor. It pays off for big groups and frequent deploys.
A share group works differently. Consumers can outnumber partitions, and each record is acknowledged on its own, with delivery attempts counted. Let's run three share consumers on a one-partition topic with six tasks. share.auto.offset.reset also defaults to latest, so we set earliest first.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic tasks --partitions 1
Created topic tasks.
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic tasks < tasks.txt
$ kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type groups --entity-name workers --add-config share.auto.offset.reset=earliest
Completed updating config for group workers.
$ (for i in 1 2 3; do kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic tasks --group workers --max-messages 2 --timeout-ms 30000 & done; wait) | sort
task-1
task-2
task-3
task-4
task-5
task-6
$ kafka-groups.sh --bootstrap-server localhost:9092 --list | sort
GROUP TYPE PROTOCOL
backfill Classic consumer
billing Classic consumer
dashboard Classic consumer
invoicing Classic consumer
last-hour Classic consumer
new-protocol Consumer consumer
rebalance-consumer Consumer consumer
rebalance-cooperative-sticky Classic consumer
rebalance-range Classic consumer
reporting Classic consumer
slow Classic consumer
workers Share shareTogether they read all six tasks, and kafka-groups.sh lists workers with type Share.
Independent jobs, like sending emails or resizing images, because there is no per-key ordering. In Python, the share consumer is still a preview.
Scenario 7: A consumer that writes to a database must not lose or double-count records when it crashes. How do you design it?
Kafka's exactly-once covers Kafka to Kafka. A database is outside it, so the design is at-least-once delivery, plus idempotent writes.
- Never lose: commit the Kafka offset only after the database transaction commits.
- Never double-count: write with an upsert keyed on a unique ID, here the payment ID, so a replay overwrites instead of adding.
Let's test it with two tables side by side: a naive insert, and the upsert. First we load six payments.
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --partitions 1
Created topic payments.
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic payments < payments.txtThe sink crashes right after the database commit of offset 3. The last Kafka commit came after offset 1, so the group resumes at 2.
$ python sink.py 3
crash after the DB commit of offset 3Now we restart the sink and check both tables.
$ python sink.py
$ python report.py
naive rows 8, total amount 950
safe rows 6, total amount 835Offsets 2 and 3 were read again. The naive table has rows 8, total amount 950: double-counted. The upsert table has rows 6, total amount 835: correct.
Store the Kafka offsets in the same database transaction as the data, and seek to them when partitions are assigned. Send records that keep failing to a dead letter topic, so one bad record doesn't block the partition.
Key points to remember
- One partition per member: partitions cap a group
- The group resumes at the committed offset: commit after processing
- auto.offset.reset only applies without a valid committed offset
- Fewer rebalances: short polls, static members, cooperative or KIP-848
- Database sinks: at-least-once + idempotent upserts
The full script
Here is a Python script that writes the files and runs every command above in order against a local Kafka 4.4; run pip install confluent-kafka first.
Show the full script that writes the files and runs every step
# ========================================================================
# Kafka Interview Q&A #3 - Consumers and consumer groups
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run on Apache Kafka 4.4.0 (KRaft mode, 1 node) with confluent-kafka 2.16.0;
# the outputs in the video and the blog post are the real outputs. To run it yourself (Linux, macOS or WSL):
# 1. Download Kafka 4.4.0 from kafka.apache.org, extract it, and put its bin folder on your PATH.
# 2. Start a local cluster (see the Kafka quickstart; node 1 listens on localhost:9092).
# 3. pip install confluent-kafka then python Kafka_Interview_QA_03_Code.py in an empty folder.
# Steps that stop or start a broker say so in a comment: do that by hand at that point.
# ========================================================================
import os, subprocess
def write(path, text):
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(text.strip("\n") + "\n")
def run(cmd, expect_error=False):
print("\n$ " + cmd, flush=True)
rc = subprocess.run(cmd, shell=True).returncode
if rc != 0 and not expect_error:
raise SystemExit("stopped: this command was expected to succeed")
# Step 0: files used by the steps
write('invoices.txt', """\
invoice-1
invoice-2
invoice-3
invoice-4
invoice-5
invoice-6
invoice-7
invoice-8
invoice-9
invoice-10
""")
write('two.py', """\
from confluent_kafka import Consumer
def member(name):
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'invoicing', 'client.id': name})
c.subscribe(['invoices'])
return c
a, b = member('consumer-a'), member('consumer-b')
for _ in range(10): # poll both while the group forms
a.poll(0.5)
b.poll(0.5)
for name, c in [('consumer-a', a), ('consumer-b', b)]:
parts = sorted(tp.partition for tp in c.assignment())
print(name, 'partitions', parts)
a.close()
b.close()
""")
write('worker.py', """\
import os, sys
from confluent_kafka import Consumer
crash_at = int(sys.argv[1]) if len(sys.argv) > 1 else -1
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'billing', 'enable.auto.commit': False,
'auto.offset.reset': 'earliest',
'session.timeout.ms': 6000}) # lab: fast detection
c.subscribe(['bills'])
done, idle = [], 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1 if c.assignment() else 0
continue
done.append(m.offset()) # process the bill
if m.offset() == crash_at:
print('processed', done, '-> crash')
os._exit(1)
if len(done) % 4 == 0: # commit every 4 records
c.commit(message=m, asynchronous=False)
if done:
c.commit(asynchronous=False) # the rest, before leaving
print('processed', done)
c.close()
""")
write('clicks.txt', """\
click-1
click-2
click-3
click-4
click-5
""")
write('reader.py', """\
import sys
from confluent_kafka import Consumer
group, reset = sys.argv[1], sys.argv[2]
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': group, 'auto.offset.reset': reset})
c.subscribe(['clicks'])
got, idle = [], 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
got.append(m.value().decode())
c.close()
print(f'group {group}, reset {reset}: read {len(got)}', got)
""")
write('rebalance.py', """\
import sys
from confluent_kafka import Consumer
mode = sys.argv[1] # range | cooperative-sticky | consumer
if mode == 'consumer':
extra = {'group.protocol': 'consumer'} # KIP-848
else:
extra = {'partition.assignment.strategy': mode}
log = {'consumer-a': [], 'consumer-b': []}
def member(name):
def note(what):
def callback(consumer, parts):
ids = sorted(p.partition for p in parts)
if ids:
log[name].append((what, ids))
return callback
c = Consumer({'bootstrap.servers': 'localhost:9092',
'client.id': name, 'group.id': 'rebalance-' + mode,
**extra})
c.subscribe(['orders'], on_assign=note('assigned'),
on_revoke=note('revoked'))
return c
a = member('consumer-a')
for _ in range(8):
a.poll(0.5)
b = member('consumer-b') # a second consumer joins
for _ in range(16):
a.poll(0.5)
b.poll(0.5)
for name, events in log.items():
print(mode, name, events)
a.close()
b.close()
""")
write('slow.py', """\
import time
from confluent_kafka import Consumer
def show(what):
return lambda c, parts: print(what, [p.partition for p in parts])
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'slow', 'auto.offset.reset': 'earliest',
'enable.auto.commit': False,
'session.timeout.ms': 6000, # lab: default 45 s
'max.poll.interval.ms': 6000}) # lab: default 5 min
c.subscribe(['bills'], on_assign=show('assigned'),
on_revoke=show('revoked'), on_lost=show('lost'))
seen = 0
while seen < 2:
m = c.poll(1.0)
if m is None:
continue
if m.error():
print('poll error:', m.error().name())
continue
seen += 1
print('processing offset', m.offset())
if seen == 1:
time.sleep(8) # slow processing
c.close()
""")
write('produce.py', """\
import sys
from confluent_kafka import Producer
p = Producer({'bootstrap.servers': 'localhost:9092'})
first, count = int(sys.argv[1]), int(sys.argv[2])
for i in range(first, first + count):
p.produce('events', value=f'event-{i}', partition=i % 3)
p.flush()
print('produced', count, 'events')
""")
write('catchup.py', """\
from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'reporting', 'auto.offset.reset': 'earliest'})
c.subscribe(['events'])
n, idle = 0, 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
n += 1
c.close() # commits the final offsets
print('reporting read', n, 'events')
""")
write('lag.py', """\
from confluent_kafka import Consumer, TopicPartition
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'reporting'})
parts = [TopicPartition('events', p) for p in range(3)]
total = 0
for tp in c.committed(parts, timeout=10):
low, high = c.get_watermark_offsets(tp, timeout=10)
lag = high - tp.offset
total += lag
print(f'partition {tp.partition}: committed {tp.offset}, '
f'end {high}, lag {lag}')
print('total lag', total)
c.close()
""")
write('kip848.py', """\
from confluent_kafka import Consumer
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'new-protocol',
'group.protocol': 'consumer', # opt in to KIP-848
'auto.offset.reset': 'earliest'})
c.subscribe(['events'])
n, idle = 0, 0
while idle < 5:
m = c.poll(1.0)
if m is None:
idle += 1
elif not m.error():
n += 1
c.close()
print('new-protocol read', n, 'events')
""")
write('tasks.txt', """\
task-1
task-2
task-3
task-4
task-5
task-6
""")
write('payments.txt', """\
pay-1,100
pay-2,250
pay-3,75
pay-4,40
pay-5,310
pay-6,60
""")
write('tables.sql', """\
create table if not exists naive (payment_id text, amount int);
create table if not exists safe (
payment_id text primary key, amount int);
""")
write('sink.py', """\
import os, sqlite3, sys
from confluent_kafka import Consumer
UPSERT = ('insert into safe values (?, ?) on conflict (payment_id) '
'do update set amount = excluded.amount')
db = sqlite3.connect('shop.db')
db.executescript(open('tables.sql').read())
crash_at = int(sys.argv[1]) if len(sys.argv) > 1 else -1
c = Consumer({'bootstrap.servers': 'localhost:9092',
'group.id': 'db-sink', 'enable.auto.commit': False,
'auto.offset.reset': 'earliest',
'session.timeout.ms': 6000}) # lab: fast detection
c.subscribe(['payments'])
while msgs := c.consume(10, timeout=10):
for m in msgs:
if m.error():
raise SystemExit(m.error())
row = m.value().decode().split(',') # payment_id, amount
with db: # one DB transaction
db.execute('insert into naive values (?, ?)', row)
db.execute(UPSERT, row)
if m.offset() == crash_at:
print('crash after the DB commit of offset', m.offset())
os._exit(1)
if m.offset() % 2 == 1: # commit Kafka every 2
c.commit(message=m, asynchronous=False)
c.close()
""")
write('report.py', """\
import sqlite3
db = sqlite3.connect('shop.db')
for table in ['naive', 'safe']:
q = f'select count(*), sum(amount) from {table}'
print(table, 'rows %d, total amount %d' % db.execute(q).fetchone())
""")
# Step 1: A topic created without a partition count gets the broker default: 1 partition
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic invoices')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic invoices')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic invoices < invoices.txt')
# Step 2: Two consumers in one group, one partition: the second one is idle
run('python two.py')
# Step 3: Add a partition and start the group again: both consumers work
run('kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic invoices --partitions 2')
run('python two.py')
# Step 4: A bills topic with ten records
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic bills --partitions 1')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic bills < invoices.txt')
# Step 5: The worker commits every 4 records and crashes at offset 6
run('python worker.py 6')
# Step 6: The committed offset is 4: the next record the group will read
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing')
run('kafka-topics.sh --bootstrap-server localhost:9092 --list | grep consumer_offsets')
# Step 7: Restart: the group resumes at offset 4, so 4, 5 and 6 are processed again
run('python worker.py')
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing')
# Step 8: Five old clicks; two new groups with latest and earliest
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic clicks --partitions 1')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic clicks < clicks.txt')
run('python reader.py dashboard latest')
run('python reader.py backfill earliest')
# Step 9: Java consumer: auto.offset.reset=by_duration:PT1H (Kafka 4.0+)
run('kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic clicks --group last-hour --command-property auto.offset.reset=by_duration:PT1H --max-messages 5')
# Step 10: A committed offset wins over auto.offset.reset
run('echo click-6 | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic clicks')
run('python reader.py backfill latest')
# Step 11: Fix the group that skipped history: reset its offsets to earliest
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group dashboard --topic clicks --to-earliest --execute')
run('python reader.py dashboard latest')
# Step 12: A second consumer joins: eager range vs cooperative-sticky vs the KIP-848 protocol
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3')
run('python rebalance.py range')
run('python rebalance.py cooperative-sticky')
run('python rebalance.py consumer')
# Step 13: Processing longer than max.poll.interval.ms: the consumer leaves the group
run('python slow.py 2> slow.log')
# Step 14: Nine events over 3 partitions; the reporting group reads them all
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic events --partitions 3')
run('python produce.py 0 9')
run('python catchup.py')
# Step 15: Reporting is down while 12 more events arrive: lag per partition
run('python produce.py 9 12')
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group reporting')
# Step 16: The same lag from code: committed offsets and watermarks
run('python lag.py')
# Step 17: A consumer that opts in to the KIP-848 protocol; the group state shows the server-side assignor
run('python kip848.py')
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group new-protocol --state')
# Step 18: A share group: three consumers on a one-partition topic share six tasks
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic tasks --partitions 1')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic tasks < tasks.txt')
run('kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type groups --entity-name workers --add-config share.auto.offset.reset=earliest')
run('(for i in 1 2 3; do kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic tasks --group workers --max-messages 2 --timeout-ms 30000 & done; wait) | sort')
run('kafka-groups.sh --bootstrap-server localhost:9092 --list | sort')
# Step 19: Six payments
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --partitions 1')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic payments < payments.txt')
# Step 20: The sink crashes after the DB commit of offset 3, before the next Kafka commit
run('python sink.py 3')
# Step 21: Restart: offsets 2 and 3 are read again; the naive table double counts, the upsert does not
run('python sink.py')
run('python report.py')
Part 4: Replication and Fault Tolerance
The sample data on this page is invented for teaching. Every terminal output is the real output of the command above it, run on Apache Kafka 4.4.0 in KRaft mode with confluent-kafka 2.16.0; topic IDs are masked. Part of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT. Features, limits and their status (GA or preview) are current as of October 2026; check the official Apache Kafka documentation (kafka.apache.org) for anything you quote in an interview.
No comments:
Post a Comment
Note: Only a member of this blog may post a comment.