Kafka Interview Questions and Answers Part 4 - Replication and Fault Tolerance

Kafka Interview Q&A · Part 4 of 10

In this post we are going to cover replication and fault tolerance: how replication works, the ISR, min.insync.replicas with acks=all, unclean leader election, racks and availability zones, and leader balance and reassignment. 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 your manager asks for a plan to survive the loss of a whole broker during peak traffic.

Replication is how Kafka keeps data safe when a broker fails: every partition is copied to several brokers. Interviewers ask about it because "a broker died, did we lose data?" is a real production question, and the honest answer depends on a few settings working together.

Part 4 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: A broker died and the team wants to know if data is safe. How does replication work in Kafka?
  4. Question 2: A partition shows fewer replicas in sync than expected. What is the ISR, and why does it shrink?
  5. Question 3: Writes fail with not enough replicas after a broker goes down. How do min insync replicas and acks all work together?
  6. Question 4: A teammate wants to enable unclean leader election to stay available. What is the risk?
  7. Question 5: The cluster spans three racks or availability zones. How do you make sure one failure does not lose data?
  8. Question 6: A broker comes back after maintenance and is overloaded. How do leader balance and partition reassignment work?
  9. Scenario 7: Your manager asks for a plan to survive the loss of a whole broker during peak traffic. What do you configure and how do you prove it?
  10. The full script

The short answer

Say this first

Each partition has replicas: one leader and followers. The ISR is the set that is caught up. With replication factor 3, min.insync.replicas=2 and acks=all, every acknowledged write is on two brokers, so one broker can fail without data loss. Keep unclean leader election off, spread replicas across racks, and prove it with a game day.

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. There are no starting files this time: each step writes the small Python script it needs. Before the first question, the full script at the end of this post also creates three topics we use later: card-swipes for the game day, clicks with replicas on brokers 2 and 1, and inventory with every replica on brokers 1 and 2.

Question 1: A broker died and the team wants to know if data is safe. How does replication work in Kafka?

Every partition is copied to several brokers. One replica is the leader: it handles the writes, and by default the reads. The followers copy from it.

Let's create payments with three partitions and three replicas each, one copy per broker.

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.
$ 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,2,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1  Elr:   LastKnownElr:
  Topic: payments  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1,2  Elr:   LastKnownElr:

The leaders are on brokers 1, 2 and 3. We also create a second topic, ledger, with two replicas and min.insync.replicas=2. We need it in Questions 2 and 3.

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

The replicas that are caught up form the ISR, the in-sync replicas. With acks=all, the leader confirms a write only when the whole ISR has it. Now python pay.py payments 6 sends six payments with an idempotent producer.

pay.py
import logging, sys
from confluent_kafka import Producer

topic, count = sys.argv[1], int(sys.argv[2])
servers = sys.argv[3] if len(sys.argv) > 3 else 'localhost:9092'
logging.basicConfig(filename='client.log')   # client logs go to a file
p = Producer({'bootstrap.servers': servers,
              'enable.idempotence': True,      # also sets acks=all
              'logger': logging.getLogger('pay')})
sent = {'delivered': 0, 'failed': 0}

def report(err, msg):
    sent['failed' if err else 'delivered'] += 1

for i in range(count):
    p.produce(topic, key=f'card-{i % 4}', value=f'payment {i}',
              on_delivery=report)
p.flush(60)
print(topic, sent)
Terminal
$ python pay.py payments 6
payments {'delivered': 6, 'failed': 0}

All six are delivered. Now broker 3 crashes: we kill it with kill -9, no clean shutdown, and describe the topic.

Now stop broker 3 (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,2  Elr:   LastKnownElr:
  Topic: payments  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,1  Elr:   LastKnownElr:
  Topic: payments  Partition: 2  Leader: 1  Replicas: 3,1,2  Isr: 1,2  Elr:   LastKnownElr:

Within seconds, the controller fences broker 3. Partition 2 now has leader 1, and every ISR is down to two brokers.

So is the data safe? Every acknowledged write is on at least two live brokers, so nothing is lost. Clients just move to the new leaders.

Real project

After a broker loss, I watch UnderReplicatedPartitions, which goes back to 0 when the broker returns, and OfflinePartitionsCount, which must stay at 0.

Question 2: A partition shows fewer replicas in sync than expected. What is the ISR, and why does it shrink?

The ISR is the leader plus the followers that are caught up. A follower drops out when it has not caught up to the leader's log end for replica.lag.time.max.ms, 30 seconds by default, or when its broker is fenced, like broker 3.

The usual causes:

  • Broker down - crash, restart, rolling upgrade
  • Slow disk - the follower can't write fast enough
  • Long GC pause - the follower stops fetching
  • Network - trouble between brokers
  • Traffic burst - more than the follower can copy

Let's list the partitions in real danger with --under-min-isr-partitions.

Terminal
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-min-isr-partitions
  Topic: ledger  Partition: 0  Leader: 1  Replicas: 1,3  Isr: 1  Elr: 3  LastKnownElr:

It shows ledger, with two replicas and only broker 1 in sync. Notice Elr: 3. That is Eligible Leader Replicas, on by default for new clusters since Kafka 4.1. When the ISR falls below min.insync.replicas, the replicas that drop out stay eligible: they have every committed record, so one can safely be elected later.

So I check the broker first, then disk and network.

Alerting

Alert on UnderReplicatedPartitions and on IsrShrinksPerSec.

Question 3: Writes fail with not enough replicas after a broker goes down. How do min insync replicas and acks all work together?

With acks=all, the leader waits for the whole current ISR. min.insync.replicas sets the floor: if the ISR is smaller, acks=all writes are rejected with NOT_ENOUGH_REPLICAS.

Let's test both topics while broker 3 is still down. First, payments has replication factor 3 and min ISR 2, and its ISR is 2.

Terminal
$ python pay.py payments 6
payments {'delivered': 6, 'failed': 0}

Six more payments are delivered. Now ledger: two replicas, min ISR 2, and an ISR of 1. acks.py sends with acks 1, then acks all with retries off, then acks all with retries on.

acks.py
from confluent_kafka import Producer

def send(**conf):
    p = Producer({'bootstrap.servers': 'localhost:9092', **conf})
    result = []
    p.produce('ledger', key='acct-7', value='debit 40',
              on_delivery=lambda err, msg: result.append(err))
    p.flush(30)
    err = result[0]
    print(conf, '->', err.name() if err else 'delivered')

send(acks='1', retries=0)
send(acks='all', retries=0)
send(acks='all', **{'message.timeout.ms': 5000})   # retries on
Terminal
$ python acks.py
{'acks': '1', 'retries': 0} -> delivered
{'acks': 'all', 'retries': 0} -> NOT_ENOUGH_REPLICAS
{'acks': 'all', 'message.timeout.ms': 5000} -> _MSG_TIMED_OUT
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic ledger
ledger:0:0

Acks 1 is accepted. Acks all with retries=0 gets NOT_ENOUGH_REPLICAS, the broker's error. With retries on, it ends as _MSG_TIMED_OUT. And that acks 1 record? kafka-get-offsets.sh says ledger:0:0. Below min ISR, the high watermark can't move, so no consumer reads it yet.

Replicas / min ISROne broker down
RF 3, min ISR 2writes go on
RF 2, min ISR 2acks=all writes fail
RF 3, min ISR 1acked by one copy

The mistake I see most is min ISR equal to the replication factor. One broker restart then blocks acks=all writes to its partitions. Three replicas with min ISR 2 survive one failure.

Set it

The broker default is min.insync.replicas=1. With ELR on, set it cluster-wide or per topic; broker-level overrides are not allowed.

Question 4: A teammate wants to enable unclean leader election to stay available. What is the risk?

Data loss. Unclean election makes a replica outside the ISR the leader, and every record it doesn't have is gone, even acknowledged ones.

Let's do it for real. clicks has two replicas, on brokers 2 and 1. We stop broker 1, and with the default min ISR of 1, broker 2 alone acknowledges three clicks.

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

Terminal
$ python pay.py clicks 3 localhost:9094
clicks {'delivered': 3, 'failed': 0}
$ kafka-topics.sh --bootstrap-server localhost:9094 --describe --topic clicks
Topic: clicks  TopicId: <topic-id>  PartitionCount: 1  ReplicationFactor: 2  Configs:
  min.insync.replicas=1
  Topic: clicks  Partition: 0  Leader: 2  Replicas: 2,1  Isr: 2  Elr:   LastKnownElr:

Then we stop broker 2 and bring broker 1 back. Its copy is behind.

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

Now start broker 1 again.

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

The partition stays offline: Leader: none. ELR points to broker 2, the replica with all the data. The safe default is to wait for it. Now we force an unclean election with kafka-leader-election.sh.

Terminal
$ kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type unclean --topic clicks --partition 0
Successfully completed leader election (UNCLEAN) for partitions clicks-0
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic clicks
Topic: clicks  TopicId: <topic-id>  PartitionCount: 1  ReplicationFactor: 2  Configs:
  min.insync.replicas=1
  Topic: clicks  Partition: 0  Leader: 1  Replicas: 2,1  Isr: 1  Elr:   LastKnownElr:
$ kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks
clicks:0:0

Broker 1 leads, and the end offset is clicks:0:0: the three clicks are gone. Now broker 2 comes back.

Now start broker 2 again.

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

It rejoins the ISR, but it truncates its log to match the new leader, and the end offset is still 0. The clicks are gone for good.

The setting unclean.leader.election.enable is false by default, and I keep it off for anything financial. If availability matters more, like for metrics, enable it on that topic only.

KRaft note

In KRaft, a periodic check applies the setting, every five minutes by default, or you run kafka-leader-election.sh.

Question 5: The cluster spans three racks or availability zones. How do you make sure one failure does not lose data?

Tell Kafka where each broker lives: set broker.rack to its rack or zone. When replicas are assigned, each partition is spread across as many racks as possible, so with three replicas, each zone holds one copy.

Let's build it up step by step:

  • Add min ISR 2 and acks=all, and every acknowledged write is in at least two zones. Losing a whole zone loses no data, and writes go on.
  • Spread the KRaft controllers too, one per zone. A majority must be alive, so three controllers survive one failure.
  • To cut cross-zone traffic, consumers can read from a replica in their own zone: client.rack on the consumer, and the rack-aware replica selector on the brokers.

Here is what that looks like in the config files.

broker-and-consumer.properties
# server.properties of the broker in zone A
broker.rack=zone-a
replica.selector.class=\
    org.apache.kafka.common.replica.RackAwareReplicaSelector

# consumer in zone A: fetch from the closest replica
client.rack=zone-a

The broker in zone A gets broker.rack=zone-a and the selector class, and a consumer in zone A sets client.rack=zone-a.

Pitfall

Partitions created before you set broker.rack keep their placement. Check them, and reassign any with two copies in one zone.

Question 6: A broker comes back after maintenance and is overloaded. How do leader balance and partition reassignment work?

A broker that comes back first catches up as a follower. Let's start broker 3 again and describe payments.

Now start broker 3 again.

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,2,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 1,2,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 2  Leader: 1  Replicas: 3,1,2  Isr: 1,2,3  Elr:   LastKnownElr:

Broker 3 is back in every ISR, but it leads nothing: partition 2 is still led by broker 1.

Each partition has a preferred leader, the first broker in its replica list. auto.leader.rebalance.enable is on by default, and the controller checks every 300 seconds. Or we trigger it ourselves.

Terminal
$ kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --topic payments --partition 2
Successfully completed leader election (PREFERRED) for partitions payments-2
$ 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,2,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 1,2,3  Elr:   LastKnownElr:
  Topic: payments  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 1,2,3  Elr:   LastKnownElr:

Broker 3 leads partition 2 again. The overload comes from catch-up: the broker copies everything it missed, and once it is back in the ISR, the next check moves its leaders back. After a long outage, throttle that catch-up, so clients keep their bandwidth.

When the load is uneven because of where replicas live, we reassign partitions. inventory has every replica on brokers 1 and 2.

Terminal
$ python pay.py inventory 30
inventory {'delivered': 30, 'failed': 0}
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic inventory
Topic: inventory  TopicId: <topic-id>  PartitionCount: 3  ReplicationFactor: 2  Configs:
  min.insync.replicas=1
  Topic: inventory  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2  Elr:   LastKnownElr:
  Topic: inventory  Partition: 1  Leader: 1  Replicas: 1,2  Isr: 1,2  Elr:   LastKnownElr:
  Topic: inventory  Partition: 2  Leader: 1  Replicas: 1,2  Isr: 1,2  Elr:   LastKnownElr:

Now we move two partitions to broker 3, throttled at 50 megabytes per second.

move.json
{"version": 1, "partitions": [
  {"topic": "inventory", "partition": 1, "replicas": [3, 1]},
  {"topic": "inventory", "partition": 2, "replicas": [2, 3]}
]}
Terminal
$ kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file move.json --execute --throttle 50000000 | grep -v log_dirs
Current partition replica assignment

Save this to use as the --reassignment-json-file option during rollback
Warning: You must run --verify periodically, until the reassignment completes, to ensure the throttle
  is removed.
The inter-broker throttle limit was set to 50000000 B/s
Successfully started partition reassignments for inventory-1,inventory-2

Then we wait until no reassignment is in progress.

Terminal
$ until kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --list | grep -q 'No partition'; do sleep 1; done

Now --verify shows both moves complete, and clears the throttles.

Terminal
$ kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file move.json --verify
Status of partition reassignment:
Reassignment of partition inventory-1 is completed.
Reassignment of partition inventory-2 is completed.

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

Partition 1 is now on brokers 3 and 1, and partition 2 on brokers 2 and 3.

Pitfall

Move a few partitions at a time, and always run --verify, or the throttle stays on.

Scenario 7: Your manager asks for a plan to survive the loss of a whole broker during peak traffic. What do you configure and how do you prove it?

First we configure for one failure:

  • 3+ brokers, RF 3, min.insync.replicas=2 - cluster-wide and on key topics
  • acks=all + idempotence - the Java default; in librdkafka you turn it on
  • Zones - replicas and controllers spread, unclean election off
  • Headroom - two brokers can carry the peak

Let's set min ISR 2 cluster-wide.

Terminal
$ kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-default --add-config min.insync.replicas=2
Completed updating default config for brokers in the cluster.
$ kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type brokers --entity-default
Default configs for brokers in the cluster are:
  min.insync.replicas=2 sensitive=false synonyms={DYNAMIC_DEFAULT_BROKER_CONFIG:min.insync.replicas=2}

It shows up as DYNAMIC_DEFAULT_BROKER_CONFIG:min.insync.replicas=2. Then we prove it with a game day. chaos.py sends three thousand card swipes over about fifteen seconds, in the background.

chaos.py
import logging, time
from confluent_kafka import Producer

logging.basicConfig(filename='client.log')
p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True,
              'logger': logging.getLogger('swipes')})
sent = {'delivered': 0, 'failed': 0}

def report(err, msg):
    sent['failed' if err else 'delivered'] += 1

for i in range(3000):              # about 15 seconds of traffic
    p.produce('card-swipes', key=f'card-{i % 50}', value=f'swipe {i}',
              on_delivery=report)
    p.poll(0)
    time.sleep(0.005)
p.flush(120)
print('sent 3000 swipes:', sent)
count.py
from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'audit',
              'auto.offset.reset': 'earliest'})
c.subscribe(['card-swipes'])
values, empty = [], 0
while empty < 5:
    m = c.poll(1.0)
    if m is None:
        empty += 1
    elif not m.error():
        values.append(m.value())
c.close()
print('read', len(values), 'swipes,', len(set(values)), 'unique')
Terminal
$ python chaos.py > chaos.out 2>&1 &
$ sleep 3

Three seconds in, we kill broker 3 with kill -9, wait for the producer, describe the topic and read everything back.

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

Terminal
$ until grep -q sent chaos.out; do sleep 1; done; cat chaos.out
sent 3000 swipes: {'delivered': 3000, 'failed': 0}
$ kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic card-swipes
Topic: card-swipes  TopicId: <topic-id>  PartitionCount: 3  ReplicationFactor: 3  Configs:
  min.insync.replicas=2
  Topic: card-swipes  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2  Elr:   LastKnownElr:
  Topic: card-swipes  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 1,2  Elr:   LastKnownElr:
  Topic: card-swipes  Partition: 2  Leader: 1  Replicas: 3,1,2  Isr: 1,2  Elr:   LastKnownElr:
$ python count.py
read 3000 swipes, 3000 unique

'delivered': 3000, 'failed': 0. The ISR is down to two, the leaders moved, and count.py reads 3000 swipes, 3000 unique: no loss, and no duplicates.

Run this drill regularly, in a test cluster first, and check these:

  • Alerts fire - UnderReplicatedPartitions, UnderMinIsrPartitionCount, OfflinePartitionsCount
  • Broker back - it rejoins the ISR
Interview tip

Say both halves: what you configure, and how you prove it. The game day is what convinces a manager.

Key points to remember

  • RF 3 + min.insync.replicas=2 + acks=all: one broker can fail
  • The ISR is who is caught up; ELR is who is still safe to elect
  • Unclean election trades acknowledged data for availability
  • broker.rack spreads replicas across zones
  • Preferred leader election, throttled reassignment, and a game day

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 #4 - Replication and fault tolerance
# 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_04_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('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic card-swipes --replica-assignment 1:2:3,2:3:1,3:1:2 --config min.insync.replicas=2')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic clicks --replica-assignment 2:1')
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic inventory --replica-assignment 1:2,1:2,1:2')

# Step 1: payments: 3 partitions, 3 replicas each, min.insync.replicas=2
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('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments')

# Step 2: ledger: replication factor 2 with min.insync.replicas=2
run('kafka-topics.sh --bootstrap-server localhost:9092 --create --topic ledger --replica-assignment 1:3 --config min.insync.replicas=2')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic ledger')

# Step 3: Send six payments with an idempotent producer (acks=all)
write('pay.py', """\
import logging, sys
from confluent_kafka import Producer

topic, count = sys.argv[1], int(sys.argv[2])
servers = sys.argv[3] if len(sys.argv) > 3 else 'localhost:9092'
logging.basicConfig(filename='client.log')   # client logs go to a file
p = Producer({'bootstrap.servers': servers,
              'enable.idempotence': True,      # also sets acks=all
              'logger': logging.getLogger('pay')})
sent = {'delivered': 0, 'failed': 0}

def report(err, msg):
    sent['failed' if err else 'delivered'] += 1

for i in range(count):
    p.produce(topic, key=f'card-{i % 4}', value=f'payment {i}',
              on_delivery=report)
p.flush(60)
print(topic, sent)
""")
run('python pay.py payments 6')

# Step 4: Broker 3 crashes (kill -9): leaders move and the ISR shrinks
# >>> now STOP broker 3 (the video stops node 3 here) <<<
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments')

# Step 5: Partitions below min.insync.replicas
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-min-isr-partitions')

# Step 6: Broker 3 is down: payments still accepts acks=all writes
run('python pay.py payments 6')

# Step 7: ledger (ISR 1, min ISR 2): acks=1 is accepted but not readable, acks=all is rejected
write('acks.py', """\
from confluent_kafka import Producer

def send(**conf):
    p = Producer({'bootstrap.servers': 'localhost:9092', **conf})
    result = []
    p.produce('ledger', key='acct-7', value='debit 40',
              on_delivery=lambda err, msg: result.append(err))
    p.flush(30)
    err = result[0]
    print(conf, '->', err.name() if err else 'delivered')

send(acks='1', retries=0)
send(acks='all', retries=0)
send(acks='all', **{'message.timeout.ms': 5000})   # retries on
""")
run('python acks.py')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic ledger')

# Step 8: Broker 3 is back: it rejoins the ISR, but leads nothing
# >>> now START broker 3 again <<<
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments')

# Step 9: Preferred leader election for payments-2
run('kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type preferred --topic payments --partition 2')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic payments')

# Step 10: inventory: every replica is on brokers 1 and 2
run('python pay.py inventory 30')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic inventory')

# Step 11: Move two inventory partitions onto broker 3, with a throttle
write('move.json', """\
{"version": 1, "partitions": [
  {"topic": "inventory", "partition": 1, "replicas": [3, 1]},
  {"topic": "inventory", "partition": 2, "replicas": [2, 3]}
]}
""")
run('kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file move.json --execute --throttle 50000000 | grep -v log_dirs')

# Step 12: Wait until no reassignment is in progress
run("until kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --list | grep -q 'No partition'; do sleep 1; done")

# Step 13: Verify the reassignment (this also removes the throttle)
run('kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file move.json --verify')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic inventory')

# Step 14: clicks (replicas 2,1): broker 1 stops, broker 2 alone acknowledges three clicks
# >>> now STOP broker 1 (the video stops node 1 here) <<<
run('python pay.py clicks 3 localhost:9094')
run('kafka-topics.sh --bootstrap-server localhost:9094 --describe --topic clicks')

# Step 15: Broker 2 stops, broker 1 comes back: the partition is offline
# >>> now STOP broker 2 (the video stops node 2 here) <<<
# >>> now START broker 1 again <<<
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic clicks')

# Step 16: Force an unclean leader election: broker 1 leads, the clicks are gone
run('kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type unclean --topic clicks --partition 0')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic clicks')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks')

# Step 17: Broker 2 comes back and truncates its log to the new leader
# >>> now START broker 2 again <<<
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic clicks')
run('kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic clicks')

# Step 18: Set min.insync.replicas=2 cluster-wide
run('kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type brokers --entity-default --add-config min.insync.replicas=2')
run('kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type brokers --entity-default')

# Step 19: Game day: start 3000 card swipes in the background
write('chaos.py', """\
import logging, time
from confluent_kafka import Producer

logging.basicConfig(filename='client.log')
p = Producer({'bootstrap.servers': 'localhost:9092',
              'enable.idempotence': True,
              'logger': logging.getLogger('swipes')})
sent = {'delivered': 0, 'failed': 0}

def report(err, msg):
    sent['failed' if err else 'delivered'] += 1

for i in range(3000):              # about 15 seconds of traffic
    p.produce('card-swipes', key=f'card-{i % 50}', value=f'swipe {i}',
              on_delivery=report)
    p.poll(0)
    time.sleep(0.005)
p.flush(120)
print('sent 3000 swipes:', sent)
""")
write('count.py', """\
from confluent_kafka import Consumer

c = Consumer({'bootstrap.servers': 'localhost:9092', 'group.id': 'audit',
              'auto.offset.reset': 'earliest'})
c.subscribe(['card-swipes'])
values, empty = [], 0
while empty < 5:
    m = c.poll(1.0)
    if m is None:
        empty += 1
    elif not m.error():
        values.append(m.value())
c.close()
print('read', len(values), 'swipes,', len(set(values)), 'unique')
""")
run('python chaos.py > chaos.out 2>&1 &')
run('sleep 3')

# Step 20: Kill broker 3 during the traffic, then check delivery, the ISR and a full read
# >>> now STOP broker 3 (the video stops node 3 here) <<<
run('until grep -q sent chaos.out; do sleep 1; done; cat chaos.out')
run('kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic card-swipes')
run('python count.py')
Next in this series

Part 5: Exactly-Once and Transactions

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)