Skip to content

Messaging, queues and Kafka internals

TL;DR

  • A queue hands each message to one worker and forgets it; pub/sub copies it to every subscriber; a log keeps it in order and lets many consumer groups replay it at their own pace.
  • Kafka's unit of ordering, parallelism and replication is the partition: pick the key and the partition count from the ordering need and the consumer count.
  • Delivery is at-least-once unless an idempotent producer meets an idempotent or transactional consumer; that pairing is what "exactly-once" means in the room.

Core concepts

Put a broker between two services when the producer must not wait for the consumer: bursts (a chat system's 70k msg/s peak against a 23k/s average), slow or failing downstreams, fan-out, replay. The price is asynchrony: the caller learns "accepted", not "done", and every consumer must tolerate duplicates.

Queue vs pub/sub vs log

A queue (SQS, a RabbitMQ queue) gives each message to one of several competing workers and deletes it on acknowledgement: easy to scale, no order across workers, no replay. Pub/sub (SNS, RabbitMQ exchanges) copies each message to every subscriber; a subscriber that is down misses it unless the broker queues per subscriber. A log (Kafka, Pulsar, Kinesis) appends records to partitions, keeps them for a retention period, and lets each consumer group track its own offset, so one topic is a queue for one group, a pub/sub bus across groups, and replayable by a new group. The costs: parallelism is capped at the partition count, and a slow record holds up everything behind it.

The log wins on throughput because it appends at the tail and reads forward: an HDD does ~150 MB/s sequentially but only ~100 random IOPS (100 x 4 KB = 400 KB/s), about 400x less, and Kafka serves reads from the page cache, so one broker sustains ~100 MB/s in and ~1 GB/s out.

Kafka anatomy: topics, partitions, offsets, consumer groups

A topic is split into partitions, each an ordered, immutable sequence of records with dense offsets. The producer picks murmur2(key) % partitions; unkeyed records go round-robin in old clients and, since 2.4, stick to one partition until the batch is full, which is round-robin per batch rather than per record. Each partition has one leader broker for all reads and writes, plus followers. A consumer group assigns every partition to exactly one member (extra members idle) and commits offsets to __consumer_offsets, a compacted internal topic, so a restarted member resumes where the group stopped.

Size partitions from the consumers: a 70k msg/s peak x 1 KB = 70 MB/s fits one broker's ingest, but at ~1k msg/s per consumer (the app-server figure) you need 70 members, hence at least 70 partitions. Take 2-3x headroom: adding partitions later remaps hash(key) % N and breaks per-key order.

One topic, three partitions, two replicas each, two independent consumer groups.

flowchart LR
    prod["Producer<br/>partition = hash(key) % 3"] --> p0
    prod --> p1
    prod --> p2
    subgraph topic["topic orders (leaders)"]
        p0[["p0: offsets 0..n"]]
        p1[["p1: offsets 0..n"]]
        p2[["p2: offsets 0..n"]]
    end
    p0 -.->|"follower fetch"| f0[["p0 replica"]]
    p1 -.->|"follower fetch"| f1[["p1 replica"]]
    p2 -.->|"follower fetch"| f2[["p2 replica"]]
    subgraph grp_a["consumer group billing"]
        ca1["member c1: p0, p1"]
        ca2["member c2: p2"]
    end
    subgraph grp_b["consumer group audit"]
        cb1["member a1: p0, p1, p2"]
    end
    p0 --> ca1
    p1 --> ca1
    p2 --> ca2
    p0 --> cb1
    p1 --> cb1
    p2 --> cb1
    ca1 -->|"commit offsets"| off[("__consumer_offsets")]
    ca2 -->|"commit offsets"| off
    cb1 -->|"commit offsets"| off

Replication: ISR, acks and the high watermark

With replication factor 3 a partition has a leader and two followers that fetch from it like consumers. The in-sync replica set (ISR) is the leader plus every follower caught up within replica.lag.time.max.ms; the high watermark marks how far every in-sync replica has caught up, and consumers never read past it, so a failover to an ISR member never exposes a record the new leader lacks. acks=0 does not wait; acks=1 waits for the leader's log and loses the record if the leader dies before followers fetch; acks=all waits for the whole ISR, gated by min.insync.replicas. RF 3 with min.insync.replicas=2 survives one dead broker without losing acknowledged writes and refuses writes (NotEnoughReplicas) with two down: consistency over availability. Keep unclean.leader.election.enable=false unless uptime is worth acknowledged data.

acks=all adds one same-datacenter round trip, ~0.5 ms per batch, amortised over hundreds of records; across regions it would add ~70 ms, so mirror asynchronously instead of stretching the ISR.

Retention, compaction and the log start offset

A partition is a chain of segment files. Time retention (retention.ms, 7 days by default) or size retention deletes whole segments from the head and advances the log start offset; a consumer whose offset fell below it gets OffsetOutOfRange and applies auto.offset.reset (earliest or latest). Capacity is rarely the limit: 100M messages/day x 1 KB = 100 GB/day, x 7 days x RF 3 = 2.1 TB, inside one server's 2-20 TB; you spread partitions for throughput, not space.

Compaction (cleanup.policy=compact) keeps the newest record per key without renumbering offsets, so the topic converges on a current-state table; a null value is a tombstone, kept for a grace period so lagging readers see the delete. Changelogs and __consumer_offsets are compacted.

Ordering, keys and rebalances

Ordering holds inside a partition and nowhere else: one key, one partition, one member at a time. Three things take it away: retries with several in-flight batches and no idempotence, changing the partition count, and salting a hot key — the only way to spread one key over several partitions, and it trades that key's order for throughput, so salt (tenant#shard) only work that needs no order.

A rebalance redistributes partitions when a member joins, leaves, misses heartbeats for session.timeout.ms or fails to poll within max.poll.interval.ms. Eager rebalancing revokes every partition, one member runs the assignor (range, round-robin, sticky) and processing stops until the new generation starts; cooperative rebalancing revokes only partitions that move, and static membership (group.instance.id) lets a member restart without a rebalance. Either way the new owner resumes from the last committed offset, so whatever the old owner processed but did not commit runs again. That window is what your consumer must make idempotent.

Delivery semantics

At-most-once: commit before processing (or acks=0) and lose records on a crash. At-least-once: acks=all with retries, commit after processing, duplicates from retries and rebalances. Effectively-once is at-least-once plus an idempotent consumer: dedupe on (topic, partition, offset) or a business id with a unique constraint or upsert (see Transactions, 2PC, sagas and idempotency).

Kafka's own exactly-once has two parts. The idempotent producer gets a producer id and numbers records per partition; the leader acknowledges a repeated sequence without appending it and rejects a gap, so retries neither duplicate nor reorder. Transactions add a transactional.id mapped to a producer id with an epoch, so a restarted instance fences its zombie predecessor; the transaction coordinator writes commit or abort markers into every partition touched; sendOffsetsToTransaction puts the consumer's offsets into the same transaction; and read_committed consumers read only up to the last stable offset. That covers loops that stay inside Kafka. The moment a side effect leaves Kafka, you are back to an idempotent sink or a transactional outbox.

Effectively-once across a lost ack and a crash: idempotent producer, acks=all, transactional consume-process-produce.

sequenceDiagram
    autonumber
    participant P as Producer (idempotent)
    participant L as Leader p2
    participant F as Follower p2 (ISR)
    participant C as Consumer (read_committed)
    participant T as Txn coordinator
    participant O as Leader out-topic
    P->>L: append seq=7 (acks=all)
    L->>F: replicate
    F-->>L: fetched, high watermark advances
    L--xP: ack lost on the network
    P->>L: retry seq=7
    L-->>P: ack with the original offset, nothing appended
    C->>L: poll below the high watermark
    L-->>C: record @42
    C->>T: begin transaction
    C->>O: produce result (txn)
    C->>T: send offsets (p2 -> 43) in the same txn
    alt consumer finishes
        T->>O: commit marker
        T-->>C: committed, read_committed readers may see it
    else consumer crashes first
        T->>O: abort marker, offset 42 stays uncommitted
        Note over C,O: the next owner of p2 replays @42 and repeats the write
    end

Retries, DLQ, poison messages, backpressure and lag

Retry transient failures with exponential backoff and jitter. Kafka has no per-message delay, so use retry topics (orders-retry-1m, orders-retry-10m) whose consumer pauses until a record is due. A poison message fails deterministically (bad schema, deleted account), and retrying it in place blocks its partition: head-of-line blocking. After N attempts publish it to a dead-letter topic with headers (source topic, partition, offset, exception, attempts), commit the offset, move on and alert on DLQ depth.

Backpressure is free with a pull model: consumers set the pace and the log absorbs the burst as lag; producers block when their buffer fills, so shed load upstream.

Lag is end offset minus committed offset per partition, and its trend beats its level. Ingest of 100M messages/day is ~1.2k/s; consumers doing 1k/s fall behind by 200/s, 17M records a day. After a one-hour outage ~4.3M records wait, and a fleet that only matches ingest never drains them: catch-up capacity must exceed arrivals, so scale the group (within the partition count) until lag trends down.

Handling a failed record without blocking the partition.

flowchart TD
    rec["record fails"] --> q_kind{"transient or<br/>deterministic?"}
    q_kind -->|"transient"| q_att{"attempts left?"}
    q_att -->|"yes"| retry["publish to retry topic<br/>with backoff and jitter"]
    retry --> commit["commit the offset,<br/>continue the partition"]
    q_att -->|"no"| dlq["publish to DLQ<br/>with source headers"]
    q_kind -->|"deterministic (poison)"| dlq
    dlq --> commit
    dlq --> alert["alert on DLQ depth,<br/>fix, replay from DLQ"]
    commit --> lag["watch lag trend<br/>per partition"]

Fan-out, priority, delay and schemas

Fan-out is Kafka's cheapest feature: a second consumer group reads the same log from the page cache. Priority is Kafka's weakest: none, so run a topic per priority and drain the urgent one first. Delayed delivery is native in Pulsar and SQS (up to 15 minutes), a plugin or TTL trick in RabbitMQ, a scheduler topic in Kafka. A schema registry stores Avro or Protobuf schemas per subject, stamps a schema id on each record and enforces compatibility (backward: upgrade consumers first; forward: upgrade producers first), so a breaking change becomes a new topic instead of an outage.

Trade-offs

System Model Ordering Replay Per-message ack, delay, priority Fan-out Best fit
Kafka partitioned log, pull per partition yes, within retention no, no, no consumer groups event streams, changelogs, pipelines
RabbitMQ AMQP broker, push with prefetch per queue, single consumer no (deleted on ack) yes, plugin or TTL, yes exchanges and bindings work queues, routing, task lifecycles
SQS standard / FIFO managed queue, pull none / per message group no visibility timeout, yes, no SNS in front serverless workers, low-ops queues
Pulsar segmented log on BookKeeper per partition or per key yes, tiered storage yes, native, no subscriptions per topic queue and stream in one
Redis Streams in-memory log per stream while in memory yes (pending list), no, no consumer groups small queues beside a cache

Choose the log when consumers are independent, when you may need replay, or when throughput is measured in hundreds of MB/s: events, activity streams, change data capture. Choose RabbitMQ when the unit of work is a task with a lifecycle: per-message acknowledgement, requeue, priorities, TTLs and routing. Choose SQS when you want no operations: visibility timeout instead of ack, dead-letter redrive and delay built in, FIFO queues for order per message group. Pulsar suits a platform team wanting queue and stream semantics on one cluster; Redis Streams is fine beside a cache you already run until the stream must outlive memory. Whatever you pick, the consumer contract is identical: idempotent processing, bounded retries, a dead-letter path and lag monitoring.

Python implementation

code/hld/mini_kafka.py collapses a cluster into one process but keeps the semantics interviews probe. A Record is immutable and a Partition never renumbers offsets, so reads bisect once compaction or retention have punched holes:

code/hld/mini_kafka.py — the partitioned log
def partition_hash(key: str) -> int:
    """Stable partitioner hash. ``hash()`` is salted per process, so two producers would
    disagree on a key's partition and per-key ordering would be lost."""
    digest = hashlib.md5(key.encode(), usedforsecurity=False).digest()
    return int.from_bytes(digest[:4], "big")


@dataclass(frozen=True, slots=True)
class Record:
    topic: str
    partition: int
    offset: int
    key: str | None
    value: str | None  # ``None`` is a tombstone: compaction deletes the key
    timestamp: float


class Partition:
    """One append-only log; offsets are never renumbered (compaction and retention leave gaps,
    so ``read`` bisects). The owning ``Broker`` serialises every call with its lock."""

    def __init__(self, topic: str, index: int) -> None:
        self.topic = topic
        self.index = index
        self._records: list[Record] = []
        self.end_offset = 0  # the offset the next record gets; for readers, the high watermark
        self.log_start_offset = 0  # oldest readable offset; retention moves it forward

    def append(self, key: str | None, value: str | None, timestamp: float) -> Record:
        record = Record(self.topic, self.index, self.end_offset, key, value, timestamp)
        self._records.append(record)
        self.end_offset += 1
        return record

    def read(self, offset: int, limit: int) -> list[Record]:
        if offset < self.log_start_offset:
            raise OffsetOutOfRangeError(f"offset {offset} is below {self.log_start_offset}")
        start = bisect.bisect_left(self._records, offset, key=lambda r: r.offset)
        return self._records[start : start + limit]

    def compact(self) -> int:
        """Keep the newest record per key; unkeyed records and tombstones stay (Kafka drops
        tombstones only after a grace period, so late readers still see the delete)."""
        newest = {r.key: r.offset for r in self._records if r.key is not None}
        keep = [r for r in self._records if r.key is None or newest[r.key] == r.offset]
        removed, self._records = len(self._records) - len(keep), keep
        return removed

    def expire(self, before: float) -> int:
        """Delete records older than ``before`` (Kafka deletes whole segments from the head)."""
        keep = [r for r in self._records if r.timestamp >= before]
        removed, self._records = len(self._records) - len(keep), keep
        self.log_start_offset = keep[0].offset if keep else self.end_offset
        return removed

The Broker owns topics, committed offsets and the idempotent-producer sequence table: append with a producer id acknowledges a repeated sequence with the original record and rejects a gap:

code/hld/mini_kafka.py — the broker
class Broker:
    """The cluster collapsed into one process. ``_lock`` guards the topic map, every
    partition's list, the committed offsets and the producer sequence table (a real broker
    locks per partition log and serves fetches from the page cache)."""

    def __init__(self, clock: Clock | None = None) -> None:
        self._clock = clock or SystemClock()
        self._topics: dict[str, list[Partition]] = {}
        self._retention: dict[str, float | None] = {}
        # (group, topic, partition) -> next offset the group will read
        self._committed: dict[tuple[str, str, int], int] = {}
        # (producer_id, topic, partition) -> (last sequence, the record it produced)
        self._sequences: dict[tuple[str, str, int], tuple[int, Record]] = {}
        self._lock = threading.Lock()

    def create_topic(self, name: str, partitions: int = 3, retention: float | None = None) -> None:
        if partitions <= 0:
            raise ValidationError("a topic needs at least one partition")
        if retention is not None and retention <= 0:
            raise ValidationError("retention must be positive")
        with self._lock:
            if name in self._topics:
                raise ConflictError(f"topic {name!r} exists")
            self._topics[name] = [Partition(name, i) for i in range(partitions)]
            self._retention[name] = retention

    def _logs(self, topic: str) -> list[Partition]:
        if topic not in self._topics:
            raise NotFoundError(f"unknown topic {topic!r}")
        return self._topics[topic]

    def _partition(self, topic: str, partition: int) -> Partition:
        logs = self._logs(topic)
        if not 0 <= partition < len(logs):
            raise ValidationError(f"topic {topic!r} has no partition {partition}")
        return logs[partition]

    def partition_count(self, topic: str) -> int:
        with self._lock:
            return len(self._logs(topic))

    def append(
        self,
        topic: str,
        partition: int,
        key: str | None,
        value: str | None,
        *,
        producer_id: str | None = None,
        sequence: int | None = None,
    ) -> Record:
        """Append one record. With a producer id, deduplicate by sequence number: a repeated
        sequence is acknowledged with the original record, a gap is rejected (Kafka's
        ``OutOfOrderSequenceException``), so the log gets neither duplicates nor holes."""
        with self._lock:
            log = self._partition(topic, partition)
            if producer_id is None:
                return log.append(key, value, self._clock.now())
            if sequence is None:
                raise ValidationError("an idempotent append needs a sequence number")
            dedup_key = (producer_id, topic, partition)
            last = self._sequences.get(dedup_key)
            if last is not None and sequence == last[0]:
                return last[1]  # the retry of a send whose ack was lost
            expected = 0 if last is None else last[0] + 1
            if sequence != expected:
                raise ConflictError(f"sequence {sequence} out of order, expected {expected}")
            record = log.append(key, value, self._clock.now())
            self._sequences[dedup_key] = (sequence, record)
            return record

    def fetch(self, topic: str, partition: int, offset: int, limit: int = 100) -> list[Record]:
        if limit <= 0:
            raise ValidationError("limit must be positive")
        with self._lock:
            return self._partition(topic, partition).read(offset, limit)

    def offsets(self, topic: str, partition: int) -> tuple[int, int]:  # (log start, end)
        with self._lock:
            log = self._partition(topic, partition)
            return log.log_start_offset, log.end_offset

    def commit(self, group: str, topic: str, partition: int, offset: int) -> None:
        """Record that ``group`` has processed everything below ``offset`` on this partition."""
        with self._lock:
            log = self._partition(topic, partition)
            if not 0 <= offset <= log.end_offset:
                raise ValidationError(f"offset {offset} is outside [0, {log.end_offset}]")
            self._committed[(group, topic, partition)] = offset

    def committed(self, group: str, topic: str, partition: int) -> int | None:
        with self._lock:
            return self._committed.get((group, topic, partition))

    def lag(self, group: str, topic: str) -> dict[int, int]:
        """Per partition: end offset minus the group's committed offset."""
        with self._lock:
            lag: dict[int, int] = {}
            for log in self._logs(topic):
                done = self._committed.get((group, topic, log.index), log.log_start_offset)
                lag[log.index] = log.end_offset - done
            return lag

    def compact(self, topic: str) -> int:
        with self._lock:
            return sum(log.compact() for log in self._logs(topic))

    def expire(self, topic: str) -> int:
        """Apply the topic's time retention: drop records older than ``now - retention``."""
        with self._lock:
            logs, retention = self._logs(topic), self._retention[topic]
            if retention is None:
                return 0
            before = self._clock.now() - retention
            return sum(log.expire(before) for log in logs)

The Producer hashes the key to a partition and numbers its sends; attempts=3 replays one send, as a client whose acks were lost would:

code/hld/mini_kafka.py — the producer
class Producer:
    """A client that partitions records and numbers them for broker-side deduplication.
    ``_lock`` guards the per-partition sequence counters and the round-robin counter."""

    def __init__(self, broker: Broker, producer_id: str) -> None:
        self._broker = broker
        self._producer_id = producer_id
        self._sequences: dict[tuple[str, int], int] = {}
        self._round_robin = itertools.count()
        self._lock = threading.Lock()

    def partition_for(self, topic: str, key: str | None) -> int:
        count = self._broker.partition_count(topic)
        if key is None:
            with self._lock:
                return next(self._round_robin) % count
        return partition_hash(key) % count

    def send(
        self, topic: str, value: str | None, key: str | None = None, attempts: int = 1
    ) -> Record:
        """Send one record. ``attempts > 1`` repeats the identical send, as a producer whose
        acks were lost would, so the caller can watch the broker drop the duplicates."""
        if attempts <= 0:
            raise ValidationError("attempts must be positive")
        partition = self.partition_for(topic, key)
        with self._lock:  # one in-flight send per partition keeps the sequence gap-free
            sequence = self._sequences.get((topic, partition), 0)
            records = [
                self._broker.append(
                    topic, partition, key, value, producer_id=self._producer_id, sequence=sequence
                )
                for _ in range(attempts)
            ]
            self._sequences[(topic, partition)] = sequence + 1
            return records[0]

ConsumerGroup runs the range assignor with an eager rebalance: every membership change revokes all partitions, bumps the generation and restarts each partition from its committed offset, which redelivers a crashed member's uncommitted records:

code/hld/mini_kafka.py — the consumer group
class ConsumerGroup:
    """Range assignment with an eager (stop-the-world) rebalance. ``_lock`` guards the members,
    the assignment and ``_positions`` (the next offset each member reads per partition);
    ``commit`` copies positions to the broker. A membership change revokes every partition and
    restarts it from its *committed* offset, so polled-but-uncommitted records are redelivered."""

    def __init__(
        self, broker: Broker, group_id: str, topic: str, auto_offset_reset: str = "earliest"
    ) -> None:
        if auto_offset_reset not in ("earliest", "latest"):
            raise ValidationError("auto_offset_reset must be 'earliest' or 'latest'")
        self._broker = broker
        self.group_id = group_id
        self.topic = topic
        self._reset = auto_offset_reset
        self._members: list[str] = []
        self._assignment: dict[str, list[int]] = {}
        self._positions: dict[tuple[str, int], int] = {}
        self.generation = 0
        self._lock = threading.Lock()

    def _rebalance(self) -> None:
        count, members = self._broker.partition_count(self.topic), sorted(self._members)
        base, extra = divmod(count, len(members) or 1)
        sizes = [base + (1 if i < extra else 0) for i in range(len(members))]  # range assignor
        bounds = list(itertools.accumulate([0, *sizes]))
        self._assignment = {m: list(range(bounds[i], bounds[i + 1])) for i, m in enumerate(members)}
        self._positions.clear()
        self.generation += 1

    def _require(self, member: str) -> list[int]:
        if member not in self._assignment:
            raise NotFoundError(f"{member!r} is not in group {self.group_id!r}")
        return self._assignment[member]

    def join(self, member: str) -> list[int]:
        with self._lock:
            if member in self._members:
                raise ConflictError(f"{member!r} is already in group {self.group_id!r}")
            self._members.append(member)
            self._rebalance()
            return list(self._assignment[member])

    def leave(self, member: str) -> None:
        """A clean leave or a missed heartbeat: the coordinator rebalances either way."""
        with self._lock:
            self._require(member)
            self._members.remove(member)
            self._rebalance()

    def assignment(self, member: str) -> list[int]:
        with self._lock:
            return list(self._require(member))

    def _start_offset(self, partition: int) -> int:
        committed = self._broker.committed(self.group_id, self.topic, partition)
        if committed is not None:
            return committed
        log_start, end = self._broker.offsets(self.topic, partition)
        return log_start if self._reset == "earliest" else end

    def poll(self, member: str, max_records: int = 100) -> list[Record]:
        """Read from each assigned partition in turn; only the in-memory position advances."""
        with self._lock:
            out: list[Record] = []
            for partition in self._require(member):
                if len(out) >= max_records:
                    break
                position = self._positions.get((member, partition))
                if position is None:
                    position = self._start_offset(partition)
                limit = max_records - len(out)
                try:
                    records = self._broker.fetch(self.topic, partition, position, limit)
                except OffsetOutOfRangeError:  # retention overtook us: apply the reset policy
                    position = self._start_offset(partition)
                    records = self._broker.fetch(self.topic, partition, position, limit)
                if records:
                    position = records[-1].offset + 1
                self._positions[(member, partition)] = position
                out.extend(records)
            return out

    def commit(self, member: str) -> dict[int, int]:
        """Commit the member's positions; returns ``{partition: committed offset}``."""
        with self._lock:
            committed: dict[int, int] = {}
            for partition in self._require(member):
                position = self._positions.get((member, partition))
                if position is not None:
                    self._broker.commit(self.group_id, self.topic, partition, position)
                    committed[partition] = position
            return committed

    def lag(self) -> dict[int, int]:
        return self._broker.lag(self.group_id, self.topic)

uv run python -m hld.mini_kafka prints:

p0: bob@0 bob@1
p1: fay@0 fay@1
p2: ann@0 ann@1 ann@2 dan@3 ann@4
idempotent send, 3 attempts: one record p0@2, log 9 -> 10
generation 2: c1=[0, 1] c2=[2]
c1 polls 5, commits {0: 3, 1: 2}
c2 polls 5 (order-0 order-3 order-5 order-6 order-8) and crashes before committing
generation 3: c1=[0, 1, 2] polls 5 again: order-0 order-3 order-5 order-6 order-8
lag before commit {0: 0, 1: 0, 2: 5}, after commit {0: 0, 1: 0, 2: 0}
compact balances: removed 3, kept ann=103@3 bob=104@4 cid=105@5
retention: expired 10, log start [3, 2, 5], new group reads order-10

Note c2's five uncommitted records returning to c1 after the rebalance, and compaction keeping one row per key at its original offset.

In the interview

Introduce the broker with its contract in one breath: "Order events go on a Kafka topic keyed by order_id, so each order's events are in order on one partition; consumers are at-least-once and idempotent, and anything that fails three times goes to a dead-letter topic."

Phrases that signal depth: "ordering is per partition, so the key is my ordering domain"; "acks=all with min.insync.replicas=2 on RF 3"; "effectively-once is at-least-once plus an idempotent consumer, so I dedupe on the business id".

How many partitions would you create for this topic?

Peak rate over per-consumer throughput: 70k msg/s at ~1k/s per consumer is 70 members, hence at least 70 partitions; I'd start near 200 because adding partitions later remaps keys.

A consumer crashes after processing half a batch. What happens?

Its partitions are reassigned after the session timeout and the new owner resumes from the last committed offset, so the processed-but-uncommitted half runs again; hence idempotent consumers and committing after processing.

How do you get exactly-once into a relational database?

Kafka transactions only cover Kafka-to-Kafka. I write the row and the consumed offset in one database transaction, or make the write idempotent with a unique event id, then commit the Kafka offset after the database commit.

One tenant generates 40% of the traffic. Now what?

Its key is one partition, so one consumer carries 40% of the load. If its events need order it gets dedicated partitions and a faster consumer; if not, I salt the key (tenant#0..k).

Why not SQS here?

SQS wins for a work queue with no operations. I choose Kafka when several teams consume the same events, when I need replay, or when order per key matters at high throughput.

Interview tip

Say "at-least-once" before the interviewer does, then immediately name the dedup point: "the consumer dedupes on the event id". Claiming exactly-once without a dedup point loses more credit than never mentioning it.

Common mistakes

  • Promising topic-level ordering: consumers see partitions interleaved. Fix: say order is per partition and choose the key to match the entity that needs it.
  • Calling Kafka exactly-once without a dedup point: retries and rebalances duplicate. Fix: idempotent producer plus idempotent consumer, or Kafka transactions for Kafka-to-Kafka only.
  • Retrying a poison message in place: the partition stalls and lag climbs for every record behind it. Fix: bounded retries via retry topics, then a DLQ with headers, commit and alert.
  • Auto-commit with asynchronous processing: the offset is committed on the next poll whether or not the handler finished, which is at-most-once. Fix: commit manually after the work is done.

Common mistake

Treating duplicates as a rare corner case. Every retry, rebalance and restart replays records. Design the consumer as idempotent from the first line, or the design fails under its most common event.

Self-check

What does acks=all guarantee, and what does min.insync.replicas add?

The leader acknowledges only after every in-sync replica has the record; min.insync.replicas=2 refuses writes with fewer than two in sync, so an acknowledged record exists on two brokers.

Why can consumers only read below the high watermark?

Records above it may exist only on the leader; if it dies and an ISR follower takes over, they vanish, so consumers must never have seen them.

How does the idempotent producer prevent duplicates?

Each producer has an id and numbers records per partition; the broker keeps the last sequence, acknowledges a repeat without appending it and rejects a gap, so retries cannot reorder.

A consumer group's lag is flat at 2M records. Is that a problem?

Flat lag means consumers match ingest with no headroom: any outage creates a backlog they can never drain. Add capacity until lag trends down.

When does compaction beat time retention?

When the topic is current state per key (balances, profiles, offsets): the latest value is kept and a new reader rebuilds state without the whole history.