Kafka Interview Questions and Answers Part 2 - Producers and Delivery Guarantees

Kafka Interview Q&A · Part 2 of 10

In this post we are going to cover producers and delivery guarantees: acks, the idempotent producer, retries and ordering, batching and compression, Python vs Java defaults, and delivery callbacks. 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 payment events must never be lost and never be duplicated by the producer.

The producer is the client that writes records to Kafka. Most "we lost events" and "we got duplicates" incidents start with its settings, so interviewers like to ask what each setting really does. They also check whether you know that the Python client and the Java client have different defaults.

Part 2 of the Real Time Kafka Interview Questions and Answers series on TechBrothersIT.

On this page

1. The short answer
2. Let's set up the cluster and some sample data
3. Question 1: Events go missing when a broker restarts. What do acks 0, 1 and all mean for a producer?
4. Question 2: A retry created duplicate messages in the topic. How does the idempotent producer prevent that?
5. Question 3: Messages for one key arrive out of order after retries. How do retries and in-flight requests affect ordering?
6. Question 4: The producer sends one message at a time and throughput is low. How do batching, linger and compression help?
7. Question 5: A teammate's Python producer behaves differently from the Java one. Which defaults should you check?
8. Question 6: A send fails and the application never notices. How do you handle delivery callbacks and errors?
9. Scenario 7: Payment events must never be lost and never duplicated by the producer. How do you configure and test it?
10. The full script

The short answer

Say this first

Use acks=all with min.insync.replicas=2 on a replication factor 3 topic, and turn on the idempotent producer (explicitly in Python, where librdkafka defaults it to off), so retries add no duplicates and keep order with up to 5 requests in flight. Tune throughput with linger.ms, batch size and compression, and always handle the delivery callback.

Let's set up the cluster and some sample data

We run a local Apache Kafka 4.4 cluster in KRaft mode with 3 nodes, and each node is both a broker and a controller. Here are the sample files: small Python producers for each question, a file of keys for the Java console producer, and a payment producer with an audit consumer for the scenario.

acks.py

from confluent_kafka import Producer

for acks in ['all', '1', '0']:   # acks 0 last: nothing races it
    p = Producer({'bootstrap.servers': 'localhost:9092', 'acks': acks})

    def report(err, m, acks=acks):
        print(f'acks {acks:3}  offset {m.offset()}  error {err}')

    p.produce('acks-demo', key='acct-7', value='pay-1', on_delivery=report)
    p.flush()

idem.py

import sys
from confluent_kafka import Producer

on = sys.argv[1] == 'on'
p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': on})
for payment in ['pay-1', 'pay-2', 'pay-2']:  # pay-2 sent twice
    p.produce('ledger', value=payment)
    p.flush()        # demo only: one batch per record

conflict.py

from confluent_kafka import Producer, KafkaException

for extra in [{'max.in.flight.requests.per.connection': 10}, {'acks': 1}]:
    try:
        Producer({'bootstrap.servers': 'localhost:9092',
                  'enable.idempotence': True, **extra})
    except KafkaException as e:
        print(extra, '->', e.args[0].str())

batch.py

from confluent_kafka import Producer

for codec in ['none', 'gzip', 'snappy', 'lz4', 'zstd']:
    p = Producer({'bootstrap.servers': 'localhost:9092',
                  'linger.ms': 1000, 'compression.type': codec})
    p.list_topics('orders-' + codec, timeout=10)   # metadata first: one batch
    for i in range(1000):
        order = f'{{"id": {i}, "store": "S-12", "amount": 25.0}}'
        # fixed timestamp, so the sizes on screen are repeatable
        p.produce('orders-' + codec, order, timestamp=1790000000000)
    p.flush()

keys.txt

c-101:java
c-606:java
c-707:java

show.properties

print.partition=true
print.key=true

parts.py

from confluent_kafka import Producer

for name in ['consistent_random', 'murmur2_random']:
    p = Producer({'bootstrap.servers': 'localhost:9092', 'partitioner': name})
    label = 'py-default' if name == 'consistent_random' else 'py-murmur2'
    for customer in ['c-101', 'c-606', 'c-707']:
        p.produce('customers', key=customer, value=label)
    p.flush()

fire.py

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True})
payouts = [('c-101', 'payout-1'), ('c-707', 'payout-2' + ' ' * 5000),
           ('c-606', 'payout-3')]
for account, payout in payouts:
    p.produce('payouts', key=account, value=payout)   # no callback
p.flush()
print('sent', len(payouts), 'payouts')

safe.py

from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True})
payouts = [('c-101', 'payout-1'), ('c-707', 'payout-2' + ' ' * 5000),
           ('c-606', 'payout-3')]
failed = []

def report(err, m):
    if err is None:
        return
    failed.append(m.key())
    kind = 'retriable' if err.retriable() else 'not retriable'
    print('FAILED', m.key().decode(), '-', err.str(), '-', kind)

for account, payout in payouts:
    p.produce('payouts', key=account, value=payout,
              on_delivery=report)
    p.poll(0)                       # serve delivery callbacks
left = p.flush(30)                  # wait for every result
if failed or left:
    raise SystemExit(f'{len(failed)} failed, {left} not delivered')

pay_producer.py

import time
from confluent_kafka import Producer

BROKERS = 'localhost:9092,localhost:9094,localhost:9096'
p = Producer({'bootstrap.servers': BROKERS,
              'enable.idempotence': True,   # explicit: off by default
              'acks': 'all'})
ok, failed = 0, []

def report(err, m):
    global ok
    if err:
        failed.append((m.value(), err))     # in production: alert
    else:
        ok += 1

for i in range(3000):
    p.produce('payments', key=f'acct-{i % 50}', value=f'pay-{i:05d}',
              on_delivery=report)
    p.poll(0)
    time.sleep(0.002)
left = p.flush(120)
print(f'sent 3000, delivered {ok}, failed {len(failed)}, '
      f'still queued {left}')

audit.py

from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'audit',
              'auto.offset.reset': 'earliest'})
c.subscribe(['payments'])
ids, idle = [], 0
while idle < 5:
    m = c.poll(1.0)
    if m is None:
        idle += 1
    elif not m.error():
        ids.append(m.value().decode())
c.close()
print(f'records in topic {len(ids)}, unique payment ids {len(set(ids))}')

Question 1: Events go missing when a broker restarts. What do acks 0, 1 and all mean for a producer?

acks tells the leader broker when to answer the producer. Each partition has one leader, and followers that copy it.

Let's go through the three values:

• acks=0: the producer does not wait for any answer. It is the fastest, but if anything fails, the record is gone and nobody knows.
• acks=1: the leader answers once it has written the record. If it dies before the followers copy it, the record is lost. That is the classic "missing events after a restart".
• acks=all: the leader answers only when all in-sync replicas have the record. This is the default in the Java client and in librdkafka.

acks	The leader answers	The record is lost if
0	never: the producer does not wait	anything fails, and nobody knows
1	when the leader has written it	the leader dies before followers copy it
all	when all in-sync replicas have it	no in-sync replica survives

Now let's create a topic with one partition and three replicas, and send the same payment with each setting using python acks.py.

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic acks-demo --replica-assignment 1:2:3
Created topic acks-demo.
$ python acks.py
acks all  offset 0  error None
acks 1    offset 1  error None
acks 0    offset None  error None

With all and 1 we get real offsets, 0 and 1. With acks 0 the delivery report says offset None, because the broker never answers.

Pitfall

acks=all waits for the in-sync replicas. If that set shrinks to just the leader, all means one. So pair it with replication factor 3 and min.insync.replicas=2. We test that in Part 4.

Question 2: A retry created duplicate messages in the topic. How does the idempotent producer prevent that?

The duplicate comes from a lost acknowledgement. The broker wrote the batch, the answer never arrived, so the producer sent it again.

Here is how idempotence fixes it, step by step. First, the producer gets a producer ID and numbers its batches with a sequence per partition. Then the broker remembers the last five batches of each producer, per partition. A retry with a sequence it already wrote is dropped, not written twice. That is also why idempotence allows at most five requests in flight.

Let's see it in the log on disk. We run python idem.py off, then python idem.py on, and read the segment with kafka-dump-log.sh. Each run sends pay-1, then pay-2 twice.

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ledger --replica-assignment 1
Created topic ledger.
$ python idem.py off
$ python idem.py on
$ kafka-dump-log.sh --print-data-log --files ../cluster/node1/ledger-0/*.log | awk '/^\| offset/ {print $2, $3, $10, $11, $14, $15}'
offset: 0 sequence: -1 payload: pay-1
offset: 1 sequence: -1 payload: pay-2
offset: 2 sequence: -1 payload: pay-2
offset: 3 sequence: 0 payload: pay-1
offset: 4 sequence: 1 payload: pay-2
offset: 5 sequence: 2 payload: pay-2

With idempotence off, offsets 0 to 2 have sequence: -1: no protection. With it on, offsets 3 to 5 get sequence 0, 1 and 2.

Now look at what it does not fix. The app sent pay-2 twice, so it is stored twice, with a new sequence. Idempotence removes the producer's own retries, within one producer session. Your own resends need a unique event ID that consumers dedupe on.

Python note

The Java client enables idempotence by default, but librdkafka, and with it the Python client, does not. Set enable.idempotence explicitly.

Question 3: Messages for one key arrive out of order after retries. How do retries and in-flight requests affect ordering?

The producer can have several requests in flight per connection. Without idempotence, batch one fails, batch two succeeds, and batch one's retry lands after it: the records are reordered.

The default Python producer is exposed here: idempotence off, and a million requests in flight.

Producer setting	After a retry
idempotence off, 1 request in flight	ordered, but one request at a time
idempotence off, more than 1 in flight	a retried batch can land after the next one
idempotence on, up to 5 in flight	ordered, and no duplicates

The old fix was one request in flight: ordered, but slow. The real fix is idempotence. The broker checks the sequence numbers, so order is kept with up to five requests in flight.

The clients enforce it. Let's run python conflict.py, which creates an idempotent producer with ten in flight, and then one with acks 1.

Terminal

$ python conflict.py
{'max.in.flight.requests.per.connection': 10} -> Failed to create producer: `max.in.flight` must be set
  <= 5 when `enable.idempotence` is true
{'acks': 1} -> Failed to create producer: `acks` must be set to `all` when `enable.idempotence` is true

Both fail as soon as we create the producer. Now the Java client: we ask the console producer for ten in flight, without asking for idempotence.

Terminal (expected error)

$ echo x | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ledger --command-property max.in.flight.requests.per.connection=10
org.apache.kafka.common.config.ConfigException: To use the idempotent producer,
  max.in.flight.requests.per.connection must be set to at most 5. Current value is 10.
  at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:121)
  at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig.java:138)
  at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerConfig.java:744)
  at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:320)
  at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:347)
  at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:332)

Java refuses it with a ConfigException, because idempotence is on by default, and since 4.0 it no longer falls back. Now we set only acks=1 on the Java console producer and check the last record on disk.

Terminal

$ echo java-acks1 | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ledger --command-property acks=1
$ kafka-dump-log.sh --print-data-log --files ../cluster/node1/ledger-0/*.log | awk '/^\| offset/ {print $2, $3, $10, $11, $14, $15}' | tail -1
offset: 6 sequence: -1 payload: java-acks1

No error this time. Java quietly turned idempotence off, and the record java-acks1 is written with sequence: -1.

Real project

For keys that need order: idempotence on, acks=all, at most five in flight, and retries left at the default.

Question 4: The producer sends one message at a time and throughput is low. How do batching, linger and compression help?

The producer groups records per partition into batches, and sends many batches per request. linger.ms is how long it waits to fill a batch, and the batch size caps its bytes.

Both clients default linger.ms to 5 milliseconds; Java changed from 0 in 4.0. But the batch size cap differs.

	Java 4.4	librdkafka
linger.ms	5	5
batch.size	16,384 bytes	1,000,000 bytes

Let's try it. python batch.py sends a thousand small orders per codec, with linger.ms set to one second for the demo. Then we read each batch header from the log.

Terminal

$ python batch.py
$ for c in none gzip snappy lz4 zstd; do kafka-dump-log.sh --files ../cluster/node1/orders-$c-0/*.log | awk '/baseOffset/ {print $5, $6, $27, $28, $31, $32}'; done
count: 1000 size: 51887 compresscodec: none
count: 1000 size: 4112 compresscodec: gzip
count: 1000 size: 8141 compresscodec: snappy
count: 1000 size: 8327 compresscodec: lz4
count: 1000 size: 2714 compresscodec: zstd

Each run became one batch of count: 1000, and the log shows its size on disk:

• none: 51887 bytes
• gzip: 4112 bytes
• snappy: 8141 bytes
• lz4: 8327 bytes
• zstd: 2714 bytes

Compression works on whole batches, so more batching means better compression, and less network and disk, for some CPU. lz4 and zstd are the usual picks.

Pitfall

The mistake I see most: calling flush after every record. That sends one record per request and kills batching. Produce in a loop, poll for callbacks, and flush only at shutdown.

Question 5: A teammate's Python producer behaves differently from the Java one. Which defaults should you check?

The Python client wraps librdkafka, and librdkafka has its own defaults. Let's put them next to the Java client.

Setting	Java client 4.4	librdkafka (Python)
enable.idempotence	true	false
acks	all	all (-1)
linger.ms	5	5
batch.size	16,384 bytes	1,000,000 bytes
delivery timeout	delivery.timeout.ms 120,000	message.timeout.ms 300,000
partitioner	murmur2	consistent_random (CRC32)

First, idempotence: on in Java, off in librdkafka. Both default acks to all, and linger to 5 milliseconds. Batch size is 16,384 bytes in Java and 1,000,000 bytes in librdkafka. The delivery timeout is two minutes in Java and five minutes in librdkafka.

The one that bites is the partitioner. Java hashes keys with murmur2, librdkafka with CRC32. Let's write the same three customers with the Java console producer and with python parts.py, once with the default partitioner and once with murmur2_random.

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --replica-assignment 1,2,3
Created topic customers.
$ kafka-console-producer.sh --bootstrap-server localhost:9092 --topic customers --reader-property parse.key=true --reader-property key.separator=: < keys.txt
$ python parts.py

Now we read the topic back with the partition and key.

Terminal

$ kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic customers --from-beginning --max-messages 9 --formatter-config show.properties | sort
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
Partition:0  c-101  py-default
Partition:0  c-606  java
Partition:0  c-606  py-murmur2
Partition:1  c-101  java
Partition:1  c-101  py-murmur2
Partition:1  c-707  java
Partition:1  c-707  py-default
Partition:1  c-707  py-murmur2
Partition:2  c-606  py-default

The first line is a banner from the 4.4 console consumer, and we can ignore it. c-101 lands in partition 1 from Java, but in partition 0 from py-default. With py-murmur2, every key matches Java. Then we dump partition 0 from disk.

Terminal

$ kafka-dump-log.sh --print-data-log --files ../cluster/node1/customers-0/*.log | awk '/^\| offset/ {print $2, $3, $10, $11, $14, $15, $16, $17}'
offset: 0 sequence: 0 key: c-606 payload: java
offset: 1 sequence: -1 key: c-101 payload: py-default
offset: 2 sequence: -1 key: c-606 payload: py-murmur2

This shows the idempotence default: Java's record has sequence: 0, and the Python records have sequence: -1.

Real project

When two languages write one topic, put the partitioner, idempotence and timeouts in a shared config, and test that a key lands in the same partition from both.

Question 6: A send fails and the application never notices. How do you handle delivery callbacks and errors?

Produce is asynchronous. It puts the record in a local queue and returns, and the result arrives later, in the delivery callback. If nobody looks at it, a failed send is silent.

Let's see it. We create payouts with max.message.bytes=1000, and python fire.py sends three payouts with no callback. Payout 2 is bigger than that cap.

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payouts --replica-assignment 1,2,3 --config max.message.bytes=1000
Created topic payouts.
$ python fire.py
sent 3 payouts
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic payouts
payouts:0:1
payouts:1:0
payouts:2:1

The script prints sent 3 payouts, but kafka-get-offsets.sh shows only two records, in partitions 0 and 2. Now python safe.py sends the same payouts with a delivery callback.

Terminal (expected error)

$ python safe.py
FAILED c-707 - Broker: Message size too large - not retriable
1 failed, 0 not delivered

The broker's error comes back: Message size too large - not retriable. The script collects failures, checks what flush left in the queue, and exits with an error: 1 failed, 0 not delivered.

Retriable errors, like a leader change, are retried for us until the delivery timeout. Whatever reaches the callback is final, so we log it, alert on it, or park the record in a dead letter topic. We never drop it.

Client tip

In Python, call poll() often so the callbacks run, and handle BufferError when the local queue is full. In Java, pass a callback to send, or check the returned future.

Scenario 7: Payment events must never be lost and never duplicated by the producer. How do you configure and test it?

First we configure it. Here is the checklist:

• Idempotence on, set explicitly because it is off by default in Python.
• acks=all, with retries left at the default, bounded by the delivery timeout.
• On the topic: replication factor 3 and min.insync.replicas=2.
• A delivery callback on every send, an alert on failures, and flush before shutdown.
• A unique payment ID in every event, so a consumer can drop resends by the app.

Then we prove it. We create payments and start pay_producer.py in the background with three thousand payments.

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --replica-assignment 1:2:3,2:3:1,3:1:2 --config min.insync.replicas=2
Created topic payments.
$ python pay_producer.py > producer.log 2> producer.err &
$ sleep 2

After two seconds, we kill broker 2 with kill -9 and describe the topic.

Now stop broker 2 (we stop that node of the cluster).

Terminal

$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments
Topic: payments  TopicId: <topic-id>  PartitionCount: 3  ReplicationFactor: 3  Configs:
  min.insync.replicas=2
  Topic: payments  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 1  Leader: 3  Replicas: 2,3,1  Isr: 3,1  Elr:   LastKnownElr:
  Topic: payments  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1  Elr:   LastKnownElr:

The controller moved broker 2's partition leadership to broker 3, and two in-sync replicas remain, so acks=all keeps working. Now we start broker 2 again, wait for the producer to finish, and run the audit consumer.

Now start broker 2 again.

Terminal

$ until grep -q delivered producer.log; do sleep 1; done; cat producer.log
sent 3000, delivered 3000, failed 0, still queued 0
$ python audit.py
records in topic 3000, unique payment ids 3000

The result: sent 3000, delivered 3000, failed 0. The audit consumer reads 3000 records with 3000 unique payment IDs. Nothing lost, nothing duplicated.

Know the limit

Idempotence covers one producer session. If the app crashes and sends again, a new producer can write a duplicate. For that, dedupe on the payment ID, or use transactions when the input is a Kafka topic.

Key points to remember

• acks=all + min.insync.replicas=2 on RF 3
• Idempotence: no duplicates from retries, order with up to 5 in flight
• Python (librdkafka): set enable.idempotence and the partitioner
• linger.ms, batch size and compression for throughput
• Always handle the delivery callback, and flush on exit

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

Full script - Kafka 4.4 running locally, pip install confluent-kafka, then run it with Python

# ========================================================================
# Kafka Interview Q&A #2 - Producers and delivery guarantees
# TechBrothersIT. Sample data is invented for teaching.
# Every step below was run on Apache Kafka 4.4.0 (KRaft mode, 3 nodes) 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_02_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('acks.py', """\
from confluent_kafka import Producer

for acks in ['all', '1', '0']:   # acks 0 last: nothing races it
    p = Producer({'bootstrap.servers': 'localhost:9092', 'acks': acks})

    def report(err, m, acks=acks):
        print(f'acks {acks:3}  offset {m.offset()}  error {err}')

    p.produce('acks-demo', key='acct-7', value='pay-1', on_delivery=report)
    p.flush()
""")
write('idem.py', """\
import sys
from confluent_kafka import Producer

on = sys.argv[1] == 'on'
p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': on})
for payment in ['pay-1', 'pay-2', 'pay-2']:  # pay-2 sent twice
    p.produce('ledger', value=payment)
    p.flush()        # demo only: one batch per record
""")
write('conflict.py', """\
from confluent_kafka import Producer, KafkaException

for extra in [{'max.in.flight.requests.per.connection': 10}, {'acks': 1}]:
    try:
        Producer({'bootstrap.servers': 'localhost:9092',
                  'enable.idempotence': True, **extra})
    except KafkaException as e:
        print(extra, '->', e.args[0].str())
""")
write('batch.py', """\
from confluent_kafka import Producer

for codec in ['none', 'gzip', 'snappy', 'lz4', 'zstd']:
    p = Producer({'bootstrap.servers': 'localhost:9092',
                  'linger.ms': 1000, 'compression.type': codec})
    p.list_topics('orders-' + codec, timeout=10)   # metadata first: one batch
    for i in range(1000):
        order = f'{{"id": {i}, "store": "S-12", "amount": 25.0}}'
        # fixed timestamp, so the sizes on screen are repeatable
        p.produce('orders-' + codec, order, timestamp=1790000000000)
    p.flush()
""")
write('keys.txt', """\
c-101:java
c-606:java
c-707:java
""")
write('show.properties', """\
print.partition=true
print.key=true
""")
write('parts.py', """\
from confluent_kafka import Producer

for name in ['consistent_random', 'murmur2_random']:
    p = Producer({'bootstrap.servers': 'localhost:9092', 'partitioner': name})
    label = 'py-default' if name == 'consistent_random' else 'py-murmur2'
    for customer in ['c-101', 'c-606', 'c-707']:
        p.produce('customers', key=customer, value=label)
    p.flush()
""")
write('fire.py', """\
from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True})
payouts = [('c-101', 'payout-1'), ('c-707', 'payout-2' + ' ' * 5000),
           ('c-606', 'payout-3')]
for account, payout in payouts:
    p.produce('payouts', key=account, value=payout)   # no callback
p.flush()
print('sent', len(payouts), 'payouts')
""")
write('safe.py', """\
from confluent_kafka import Producer

p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True})
payouts = [('c-101', 'payout-1'), ('c-707', 'payout-2' + ' ' * 5000),
           ('c-606', 'payout-3')]
failed = []

def report(err, m):
    if err is None:
        return
    failed.append(m.key())
    kind = 'retriable' if err.retriable() else 'not retriable'
    print('FAILED', m.key().decode(), '-', err.str(), '-', kind)

for account, payout in payouts:
    p.produce('payouts', key=account, value=payout,
              on_delivery=report)
    p.poll(0)                       # serve delivery callbacks
left = p.flush(30)                  # wait for every result
if failed or left:
    raise SystemExit(f'{len(failed)} failed, {left} not delivered')
""")
write('pay_producer.py', """\
import time
from confluent_kafka import Producer

BROKERS = 'localhost:9092,localhost:9094,localhost:9096'
p = Producer({'bootstrap.servers': BROKERS,
              'enable.idempotence': True,   # explicit: off by default
              'acks': 'all'})
ok, failed = 0, []

def report(err, m):
    global ok
    if err:
        failed.append((m.value(), err))     # in production: alert
    else:
        ok += 1

for i in range(3000):
    p.produce('payments', key=f'acct-{i % 50}', value=f'pay-{i:05d}',
              on_delivery=report)
    p.poll(0)
    time.sleep(0.002)
left = p.flush(120)
print(f'sent 3000, delivered {ok}, failed {len(failed)}, '
      f'still queued {left}')
""")
write('audit.py', """\
from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'audit',
              'auto.offset.reset': 'earliest'})
c.subscribe(['payments'])
ids, idle = [], 0
while idle < 5:
    m = c.poll(1.0)
    if m is None:
        idle += 1
    elif not m.error():
        ids.append(m.value().decode())
c.close()
print(f'records in topic {len(ids)}, unique payment ids {len(set(ids))}')
""")
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-none --replica-assignment 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-gzip --replica-assignment 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-snappy --replica-assignment 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-lz4 --replica-assignment 1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders-zstd --replica-assignment 1')

# Step 1: One partition with three replicas; the same payment with acks 0, 1 and all
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic acks-demo --replica-assignment 1:2:3')
run('python acks.py')

# Step 2: Idempotence off, then on: the records on disk carry sequence numbers only when it is on
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ledger --replica-assignment 1')
run('python idem.py off')
run('python idem.py on')
run("kafka-dump-log.sh --print-data-log --files ../cluster/node1/ledger-0/*.log | awk '/^\\| offset/ {print $2, $3, $10, $11, $14, $15}'")

# Step 3: librdkafka refuses idempotence with more than 5 requests in flight, or with acks 1
run('python conflict.py')

# Step 4: Java: 10 in flight is refused even without asking for idempotence (it is on by default)
run('echo x | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ledger --command-property max.in.flight.requests.per.connection=10', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 5: Java: acks=1 silently turns idempotence off (the record has no sequence number)
run('echo java-acks1 | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic ledger --command-property acks=1')
run("kafka-dump-log.sh --print-data-log --files ../cluster/node1/ledger-0/*.log | awk '/^\\| offset/ {print $2, $3, $10, $11, $14, $15}' | tail -1")

# Step 6: A thousand small orders per codec with linger.ms 1000: one batch each, and its size in the log
run('python batch.py')
run("for c in none gzip snappy lz4 zstd; do kafka-dump-log.sh --files ../cluster/node1/orders-$c-0/*.log | awk '/baseOffset/ {print $5, $6, $27, $28, $31, $32}'; done")

# Step 7: The Java console producer and the Python producer write the same keys
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --replica-assignment 1,2,3')
run('kafka-console-producer.sh --bootstrap-server localhost:9092 --topic customers --reader-property parse.key=true --reader-property key.separator=: < keys.txt')
run('python parts.py')

# Step 8: Where each key landed, per producer
run('kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic customers --from-beginning --max-messages 9 --formatter-config show.properties | sort')

# Step 9: Partition 0 on disk: the Java record has a sequence number, the Python ones do not
run("kafka-dump-log.sh --print-data-log --files ../cluster/node1/customers-0/*.log | awk '/^\\| offset/ {print $2, $3, $10, $11, $14, $15, $16, $17}'")

# Step 10: Fire and forget: the script says sent 3, the topic has 2 (max.message.bytes is 1000)
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payouts --replica-assignment 1,2,3 --config max.message.bytes=1000')
run('python fire.py')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic payouts')

# Step 11: With a delivery callback, the failure is reported and the script fails
run('python safe.py', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 12: Payments topic: 3 replicas, min.insync.replicas 2; start 3000 payments in the background
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic payments --replica-assignment 1:2:3,2:3:1,3:1:2 --config min.insync.replicas=2')
run('python pay_producer.py > producer.log 2> producer.err &')
run('sleep 2')

# Step 13: kill -9 broker 2 while the producer runs; the controller moves its partitions
# >>> now STOP broker 2 (the video stops node 2 here) <<<
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments')

# Step 14: Start broker 2 again; wait for the producer, then audit the topic
# >>> now START broker 2 again <<<
run('until grep -q delivered producer.log; do sleep 1; done; cat producer.log')
run('python audit.py')

Next in this series

Part 3: Consumers and Consumer Groups

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.