Kafka Interview Questions and Answers Part 6 - Retention, Compaction and Storage

Kafka Interview Q&A · Part 6 of 10

In this post we are going to cover retention, compaction and storage in Kafka. 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: keeping seven years of order events for audit without filling the brokers.

Kafka does not keep data forever by default. Every topic has a cleanup policy that decides when old records go away: delete them by age or size, or compact them down to the latest value per key. Interviewers ask about this because full disks, lost replay data and deleted customers that come back are real problems on real projects.

Part 6 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: The disk on the brokers is filling up. How do retention by time and by size work?
  4. Question 2: A teammate wants the latest value per customer kept forever. How does log compaction work?
  5. Question 3: Deleted customers still appear when a new service reads the compacted topic. How do tombstones work?
  6. Question 4: Old data is needed for a reprocess but retention already deleted it. How do you plan retention for replays?
  7. Question 5: Your manager asks about tiered storage to cut disk costs. What is it, and what are its limits?
  8. Question 6: A topic has thousands of tiny segment files. How do segments and their settings affect the brokers?
  9. Scenario 7: You must keep seven years of order events for audit without filling the brokers. What is your design?
  10. The full script

The short answer

Say this first

Retention deletes whole closed segments by time (retention.ms, 7 days) or size (retention.bytes, per partition). Compaction keeps the latest record per key, and a tombstone (null value) deletes a key. Plan retention from the replay window, and keep long history in tiered storage or the data lake.

Let's set up the cluster and some sample data

We use a local Kafka 4.4 cluster in KRaft mode with 1 broker. Each step writes its own small Python script with confluent-kafka, so the files show up right where we use them.

Question 1: The disk on the brokers is filling up. How do retention by time and by size work?

Kafka deletes whole segments, not single records. Each partition's log is a series of segment files, and only the newest one, the active segment, takes writes.

Let's create a clicks topic that keeps thirty days and starts a new segment every day. Then we run clicks.py, which writes twelve clicks: three each from twenty, twelve and five days ago, and three from today.

clicks.py
import time
from confluent_kafka import Producer

DAY = 24 * 3600 * 1000                 # one day in milliseconds
now = int(time.time() * 1000)
p = Producer({'bootstrap.servers': 'localhost:9092'})
for days_ago in [20, 12, 5, 0]:
    for page in ['/home', '/cart', '/pay']:
        p.produce('clicks', key='user-7', value=page,
                  timestamp=now - days_ago * DAY)
        p.flush()                      # one batch per click
print('sent 12 clicks: 3 each from 20, 12, 5 and 0 days ago')
Terminal
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic clicks --partitions 1 --config retention.ms=2592000000 --config segment.ms=86400000
Created topic clicks.
$ python clicks.py
sent 12 clicks: 3 each from 20, 12, 5 and 0 days ago
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest
clicks:0:0
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time latest
clicks:0:12

The earliest offset is 0 and the latest is 12. That's four segments, offsets 0 to 11.

Now let's set retention.ms to seven days, the default.

Terminal
$ kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config retention.ms=604800000
Completed updating config for topic clicks.

Then we wait for the next retention check by polling the earliest offset.

Terminal
$ until kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest | grep -q ':6$'; do sleep 1; done
Terminal
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest
clicks:0:6
$ ls kafka-logs/clicks-0 | grep '\.log'
00000000000000000000.log.deleted
00000000000000000003.log.deleted
00000000000000000006.log
00000000000000000009.log

The two segments older than seven days are gone, and the earliest offset is now 6. A segment goes when its newest record is older than the retention. Its files are renamed with .deleted, as the listing shows, and removed a minute later.

Next, retention.bytes caps the size of each partition, not the topic. Let's set it to 200 bytes and wait again.

Terminal
$ kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config retention.bytes=200
Completed updating config for topic clicks.
Terminal
$ until kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest | grep -q ':9$'; do sleep 1; done
Terminal
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest
clicks:0:9

The oldest segment left goes too, and the earliest offset is 9. Whichever limit is hit first wins.

Lab only

This lab checks retention every second. The default is every five minutes. On a real cluster, lower retention on the biggest topics to buy time, then size the disks for the retention you promised.

Question 2: A teammate wants the latest value per customer kept forever. How does log compaction work?

We set cleanup.policy=compact. Nothing is deleted by age. Instead, the log cleaner keeps at least the latest record for each key, and removes older records with the same key.

Let's create a customers topic with compaction and a segment.ms of one hour. We run customers.py, which sends seven changes: c1 moves from Austin to Chicago, c2 from Boston to Edison, c3 is Denver, then c2 is deleted. c4 arrives two hours later, so it starts a new segment.

customers.py
from confluent_kafka import Producer

T0 = 1793007000000                     # 2026-10-26 09:30 UTC
p = Producer({'bootstrap.servers': 'localhost:9092'})
changes = [('c1', '{"city": "Austin"}'), ('c2', '{"city": "Boston"}'),
           ('c1', '{"city": "Chicago"}'), ('c3', '{"city": "Denver"}'),
           ('c2', '{"city": "Edison"}'), ('c2', None)]   # None: delete
for i, (key, value) in enumerate(changes):
    p.produce('customers', key=key, value=value, timestamp=T0 + i * 60000)
    p.flush()
# two hours later: starts a new segment (segment.ms is one hour)
p.produce('customers', key='c4', value='{"city": "Fresno"}',
          timestamp=T0 + 2 * 3600000)
p.flush()
print('sent 7 customer changes')
Terminal
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --partitions 1 --config cleanup.policy=compact --config segment.ms=3600000
Created topic customers.
$ python customers.py
sent 7 customer changes

The cleaner only works on closed segments, never the active one. Now we run compacted.py, which waits until the cleaner has run and then reads the whole topic.

compacted.py
import time
from confluent_kafka import OFFSET_BEGINNING, Consumer, TopicPartition

def read_all(topic):
    c = Consumer({'bootstrap.servers': 'localhost:9092',
                  'group.id': 'peek', 'enable.auto.commit': False})
    high = c.get_watermark_offsets(TopicPartition(topic, 0))[1]
    c.assign([TopicPartition(topic, 0, OFFSET_BEGINNING)])
    rows = []
    while not rows or rows[-1][0] < high - 1:
        m = c.poll(5.0)
        if m is None:
            break
        rows.append((m.offset(), m.key().decode(), m.value()))
    c.close()
    return rows

rows, t0 = read_all('customers'), time.time()
while len(rows) == 7 and time.time() - t0 < 120:   # wait for the cleaner
    time.sleep(1)
    rows = read_all('customers')
for offset, key, value in rows:
    print('offset', offset, 'key', key, 'value', value and value.decode())
Terminal
$ python compacted.py
offset 2 key c1 value {"city": "Chicago"}
offset 3 key c3 value {"city": "Denver"}
offset 5 key c2 value None
offset 6 key c4 value {"city": "Fresno"}

What's left: c1 Chicago at offset 2, c3 Denver at 3, the c2 delete at 5, and c4 Fresno at 6. Let's dump the cleaned segment with kafka-dump-log.sh.

Terminal
$ kafka-dump-log.sh --files kafka-logs/customers-0/00000000000000000000.log --print-data-log 2>/dev/null | grep '| offset'
| offset: 2 CreateTime: 1793007120000 keySize: 2 valueSize: 19 sequence: -1 headerKeys: [] key: c1
  payload: {"city": "Chicago"}
| offset: 3 CreateTime: 1793007180000 keySize: 2 valueSize: 18 sequence: -1 headerKeys: [] key: c3
  payload: {"city": "Denver"}
| offset: 5 CreateTime: 1793007300000 keySize: 2 valueSize: -1 sequence: -1 headerKeys: [] key: c2

It shows offsets 2, 3 and 5. Offsets never change and order is kept, so compaction leaves gaps.

Cleaning starts when the dirty part of the log reaches min.cleanable.dirty.ratio, 0.5 by default. min.compaction.lag.ms sets a minimum age, and max.compaction.lag.ms sets a deadline.

Remember

Compaction is eventual. Readers can still see old values for a while.

Question 3: Deleted customers still appear when a new service reads the compacted topic. How do tombstones work?

A delete in a compacted topic is a tombstone: a record with the key and a null value. The cleaner removes the older values for that key. In the dump from Question 2, the c2 tombstone at offset 5 has valueSize: -1, so it has no value at all.

A new service that reads from the start must apply the tombstone as a delete. Let's run rebuild.py, which builds a customer table and removes a key when the value is null.

rebuild.py
from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092',
              'group.id': 'customer-service',
              'auto.offset.reset': 'earliest'})
c.subscribe(['customers'])
table, empty = {}, 0
while empty < 3:
    m = c.poll(1.0)
    if m is None:
        empty += 1
        continue
    key = m.key().decode()
    if m.value() is None:              # tombstone: delete the customer
        table.pop(key, None)
    else:
        table[key] = m.value().decode()
c.close()
for key in sorted(table):
    print(key, table[key])
Terminal
$ python rebuild.py
c1 {"city": "Chicago"}
c3 {"city": "Denver"}
c4 {"city": "Fresno"}

Its table has c1, c3 and c4. c2 is gone, as it should be.

So when deleted customers still appear, we check three things:

  • A real tombstone? The source must send the key with a null value, not a deleted flag inside the value.
  • Null handled? A consumer that skips null values never deletes.
  • Same partition? By default, the Java client (murmur2) and librdkafka (CRC32) can hash the same key to different partitions, so a Java and a Python producer can disagree.

The tombstone itself is removed after delete.retention.ms, 24 hours by default.

Pitfall

A consumer that is offline longer than delete.retention.ms can miss the delete. Set it longer than your longest outage.

Question 4: Old data is needed for a reprocess but retention already deleted it. How do you plan retention for replays?

Once a segment is deleted, Kafka can't bring it back. Raising retention now only protects what is still on disk.

Let's see it on the clicks topic. A replay group asks to reset to October 1, and we use --dry-run to see where it would start.

Terminal
$ kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay --topic clicks --reset-offsets --to-datetime 2026-10-01T00:00:00.000 --dry-run
GROUP           TOPIC           PARTITION  NEW-OFFSET
replay          clicks          0          9

The new offset is 9, the earliest offset left. Everything before it is gone.

So we plan retention from the replay window: how long it takes to notice a bug, fix it and rerun, plus a margin. If bugs are found two weeks later, seven days is not enough. Then we size the disks.

How to size it
Replay windowtime to notice a bug, fix it and rerun, plus a margin
Diskbytes per day × days kept × replication factor + headroom
retention.bytesper partition: × partitions for the topic

For long windows, don't keep everything on broker disks. Use tiered storage, or archive the topic to the data lake with a sink connector, and replay from there.

Tip

Retention is per topic: hours for noisy logs, weeks for business events. A retention.ms of -1, or an empty cleanup.policy since Kafka 4.2, keeps data forever.

Question 5: Your manager asks about tiered storage to cut disk costs. What is it, and what are its limits?

Tiered storage keeps recent segments on the broker disks and copies closed segments to remote storage, like S3 or HDFS. Consumers still read old offsets through the brokers.

It's production-ready since Kafka 3.9. First we turn it on in the broker config, then per topic with remote.storage.enable. local.retention.ms sets how long segments stay on local disk, and retention.ms sets how long they're kept at all.

tiered-storage.properties
# broker (server.properties)
remote.log.storage.system.enable=true
remote.log.storage.manager.class.name=com.example.S3StorageManager
remote.log.storage.manager.class.path=/opt/kafka/rsm-plugin/*
remote.log.metadata.manager.listener.name=PLAINTEXT

# topic: one day on local disk, one year in total
remote.storage.enable=true
local.retention.ms=86400000
retention.ms=31536000000

The catch: Apache Kafka ships no remote storage manager for S3 or any cloud store, so we need a plugin. And without the broker setting, the topic setting is rejected. Let's try it on our broker, where tiered storage is off.

Terminal (expected error)
$ kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config remote.storage.enable=true
Error while executing config command with args '--bootstrap-server localhost:9092 --entity-type topics
  --alter --entity-name clicks --add-config remote.storage.enable=true'
java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.InvalidConfigurationException:
  Tiered Storage functionality is disabled in the broker. Topic cannot be configured with remote log
    storage.
  at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
  at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2096)
Caused by: org.apache.kafka.common.errors.InvalidConfigurationException: Tiered Storage functionality
  is disabled in the broker. Topic cannot be configured with remote log storage.

The error says "Tiered Storage functionality is disabled in the broker".

The limits to know:

  • No compacted topics.
  • To turn it off at the broker, we first disable it on every topic.
  • Reads from remote storage are slower than local ones.
  • Kafka 4.4 adds delayed upload, with remote.copy.lag.ms and remote.copy.lag.bytes.
Tip

It cuts broker disk and makes long retention cheaper. Test the plugin well, and keep the brokers on current patch releases.

Question 6: A topic has thousands of tiny segment files. How do segments and their settings affect the brokers?

Every segment is a set of files: the log, an offset index, a time index, and usually a producer snapshot.

Let's create a scans topic with a segment.ms of one minute. Then scans.py writes five scans stamped two minutes apart, so each scan starts a new segment.

scans.py
import time
from confluent_kafka import Producer

T0 = int(time.time() * 1000) - 600000  # ten minutes ago
p = Producer({'bootstrap.servers': 'localhost:9092'})
for i in range(5):                     # one pallet scan every 2 minutes
    p.produce('scans', key='dock-1', value=f'pallet {i}',
              timestamp=T0 + i * 120000)
    p.flush()
print('sent 5 scans')
Terminal
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic scans --partitions 1 --config segment.ms=60000
Created topic scans.
$ python scans.py
sent 5 scans

Now let's list the partition folder with sizes.

Terminal
$ ls -l kafka-logs/scans-0 | awk 'NR > 1 {print $5, $9}'
0 00000000000000000000.index
82 00000000000000000000.log
12 00000000000000000000.timeindex
0 00000000000000000001.index
82 00000000000000000001.log
10 00000000000000000001.snapshot
12 00000000000000000001.timeindex
0 00000000000000000002.index
82 00000000000000000002.log
10 00000000000000000002.snapshot
12 00000000000000000002.timeindex
0 00000000000000000003.index
82 00000000000000000003.log
10 00000000000000000003.snapshot
12 00000000000000000003.timeindex
10485760 00000000000000000004.index
82 00000000000000000004.log
10 00000000000000000004.snapshot
10485756 00000000000000000004.timeindex
8 leader-epoch-checkpoint
43 partition.metadata

That's five segments for five records. Each log file is 82 bytes. The closed segments' indexes are trimmed, but the active segment's indexes are preallocated at ten mebibytes (10485760 bytes), from segment.index.bytes. Let's check the segment settings of the topic.

Terminal
$ kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --describe --entity-name scans --all | grep -E '^  segment.(bytes|index.bytes|ms)='
  segment.bytes=1073741824 sensitive=false synonyms={DEFAULT_CONFIG:log.segment.bytes=1073741824}
  segment.index.bytes=10485760 sensitive=false
    synonyms={DEFAULT_CONFIG:log.index.size.max.bytes=10485760}
  segment.ms=60000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:segment.ms=60000}

segment.bytes is 1073741824 (one gibibyte) and segment.ms is our 60000. Now multiply this by thousands:

  • Open files: every segment has a log and indexes.
  • Memory maps: two per segment for its indexes, limited by vm.max_map_count.
  • Startup and recovery get slower.
  • Retention and the cleaner have more files to work through.

The defaults roll a segment at one gibibyte or seven days, whichever comes first. Tiny segments usually come from a small segment.ms, set to make compaction or retention kick in sooner. Since Kafka 4.0, segment.bytes can't go below one mebibyte.

Fix

Raise segment.ms back up, and on the brokers check the open file limit and vm.max_map_count.

Scenario 7: You must keep seven years of order events for audit without filling the brokers. What is your design?

Kafka is the transport, not the archive. Let's walk through the design step by step.

First, orders stay on the brokers for the replay window, here fourteen days, with replication factor three in production. Let's create the orders topic with 14 days of retention.

Terminal
$ kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --config retention.ms=1209600000
Created topic orders.
$ kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --describe --entity-name orders
Dynamic configs for topic orders are:
  retention.ms=1209600000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=1209600000}

The dynamic config shows retention.ms=1209600000, which is 14 days.

Then a sink connector writes every event to object storage, in Parquet or a table format like Iceberg, partitioned by day. The bucket has a seven-year immutable retention policy, so nobody can change or delete the audit trail.

Next, we keep the schema with the data. We register every schema version and store it with the files, so a file from year one is still readable in year seven.

Tiered storage is the other option. Let's compare the two.

HowTrade-off
Tiered storageretention.ms 7 years, local.retention.ms a few daysreads via brokers; topic can be deleted
Lake archivesink connector to object storagequery and replay without Kafka

Tiered storage is simpler for replays, but every read still goes through the brokers, the archive is tied to Kafka, and an admin can still delete the topic.

Prove it

A daily job compares each day's event count in the lake with the topic, before retention deletes it. Auditors query the lake, never the brokers.

Key points to remember

  • Retention deletes whole closed segments, by time or size
  • retention.bytes is per partition; the check runs every 5 minutes
  • Compaction: the latest record per key; offsets never change
  • Tombstones: key + null value, kept for delete.retention.ms
  • Long history: tiered storage or the lake; keep segments large

The full script

This is a Python script that writes the files and runs every command 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 #6 - Retention, compaction and storage
# 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_06_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")

run('ln -s ../cluster/node1 kafka-logs   # kafka-logs = the broker log.dirs folder')

# Step 1: clicks: 30 days of retention, a new segment every day, 12 clicks over 20 days
write('clicks.py', """\
import time
from confluent_kafka import Producer

DAY = 24 * 3600 * 1000                 # one day in milliseconds
now = int(time.time() * 1000)
p = Producer({'bootstrap.servers': 'localhost:9092'})
for days_ago in [20, 12, 5, 0]:
    for page in ['/home', '/cart', '/pay']:
        p.produce('clicks', key='user-7', value=page,
                  timestamp=now - days_ago * DAY)
        p.flush()                      # one batch per click
print('sent 12 clicks: 3 each from 20, 12, 5 and 0 days ago')
""")
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic clicks --partitions 1 --config retention.ms=2592000000 --config segment.ms=86400000')
run('python clicks.py')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time latest')

# Step 2: Lower retention.ms to 7 days (the default)
run('kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config retention.ms=604800000')

# Step 3: Wait for the next retention check (poll the earliest offset)
run("until kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest | grep -q ':6$'; do sleep 1; done")

# Step 4: After the retention check: two segments deleted
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest')
run("ls kafka-logs/clicks-0 | grep '\\.log'")

# Step 5: Cap each partition at 200 bytes with retention.bytes
run('kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config retention.bytes=200')

# Step 6: Wait for the next retention check (poll the earliest offset)
run("until kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest | grep -q ':9$'; do sleep 1; done")

# Step 7: After the retention check: one more segment deleted
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks --time earliest')

# Step 8: A replay from October 1 can only start at the earliest offset left
run('kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group replay --topic clicks --reset-offsets --to-datetime 2026-10-01T00:00:00.000 --dry-run')

# Step 9: Tiered storage is off on this broker: the topic setting is rejected
run('kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --alter --entity-name clicks --add-config remote.storage.enable=true', expect_error=True)   # EXPECTED TO FAIL: the video shows this error

# Step 10: customers: a compacted topic with seven changes
write('customers.py', """\
from confluent_kafka import Producer

T0 = 1793007000000                     # 2026-10-26 09:30 UTC
p = Producer({'bootstrap.servers': 'localhost:9092'})
changes = [('c1', '{"city": "Austin"}'), ('c2', '{"city": "Boston"}'),
           ('c1', '{"city": "Chicago"}'), ('c3', '{"city": "Denver"}'),
           ('c2', '{"city": "Edison"}'), ('c2', None)]   # None: delete
for i, (key, value) in enumerate(changes):
    p.produce('customers', key=key, value=value, timestamp=T0 + i * 60000)
    p.flush()
# two hours later: starts a new segment (segment.ms is one hour)
p.produce('customers', key='c4', value='{"city": "Fresno"}',
          timestamp=T0 + 2 * 3600000)
p.flush()
print('sent 7 customer changes')
""")
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic customers --partitions 1 --config cleanup.policy=compact --config segment.ms=3600000')
run('python customers.py')

# Step 11: Wait for the log cleaner, then read the whole topic
write('compacted.py', """\
import time
from confluent_kafka import OFFSET_BEGINNING, Consumer, TopicPartition

def read_all(topic):
    c = Consumer({'bootstrap.servers': 'localhost:9092',
                  'group.id': 'peek', 'enable.auto.commit': False})
    high = c.get_watermark_offsets(TopicPartition(topic, 0))[1]
    c.assign([TopicPartition(topic, 0, OFFSET_BEGINNING)])
    rows = []
    while not rows or rows[-1][0] < high - 1:
        m = c.poll(5.0)
        if m is None:
            break
        rows.append((m.offset(), m.key().decode(), m.value()))
    c.close()
    return rows

rows, t0 = read_all('customers'), time.time()
while len(rows) == 7 and time.time() - t0 < 120:   # wait for the cleaner
    time.sleep(1)
    rows = read_all('customers')
for offset, key, value in rows:
    print('offset', offset, 'key', key, 'value', value and value.decode())
""")
run('python compacted.py')

# Step 12: kafka-dump-log.sh: the cleaned segment keeps the original offsets
run("kafka-dump-log.sh --files kafka-logs/customers-0/00000000000000000000.log --print-data-log 2>/dev/null | grep '| offset'")

# Step 13: A new service rebuilds its customer table and applies the tombstone
write('rebuild.py', """\
from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092',
              'group.id': 'customer-service',
              'auto.offset.reset': 'earliest'})
c.subscribe(['customers'])
table, empty = {}, 0
while empty < 3:
    m = c.poll(1.0)
    if m is None:
        empty += 1
        continue
    key = m.key().decode()
    if m.value() is None:              # tombstone: delete the customer
        table.pop(key, None)
    else:
        table[key] = m.value().decode()
c.close()
for key in sorted(table):
    print(key, table[key])
""")
run('python rebuild.py')

# Step 14: scans: segment.ms of one minute, five scans two minutes apart
write('scans.py', """\
import time
from confluent_kafka import Producer

T0 = int(time.time() * 1000) - 600000  # ten minutes ago
p = Producer({'bootstrap.servers': 'localhost:9092'})
for i in range(5):                     # one pallet scan every 2 minutes
    p.produce('scans', key='dock-1', value=f'pallet {i}',
              timestamp=T0 + i * 120000)
    p.flush()
print('sent 5 scans')
""")
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic scans --partitions 1 --config segment.ms=60000')
run('python scans.py')

# Step 15: Five segments for five records
run("ls -l kafka-logs/scans-0 | awk 'NR > 1 {print $5, $9}'")

# Step 16: The segment settings of the topic
run("kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --describe --entity-name scans --all | grep -E '^  segment.(bytes|index.bytes|ms)='")

# Step 17: orders: 14 days on the brokers for replays
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --config retention.ms=1209600000')
run('kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --describe --entity-name orders')
Next in this series

Part 7: Kafka Connect, CDC and Schemas

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.

Subscribe to: Post Comments (Atom)