Apache Kafka Interview Questions (2026): Must-Know Q&A

A broker is a single Kafka server that stores data and serves reads and writes. A topic is a named category of messages, and Kafka splits every topic into one or more partitions, each an ordered, append-only log. Producers write to partitions, consumers read from them, and since Kafka moved to KRaft, a small quorum of controller nodes tracks cluster metadata directly inside Kafka instead of relying on a separate ZooKeeper ensemble.

Partitions are the unit that actually gets distributed across brokers. A topic with 12 partitions and a cluster of 3 brokers means each broker holds roughly a third of that topic's data as leader, plus replicas of the other two-thirds.

If the message has a key, Kafka hashes it (murmur2 by default) and mods that hash by the partition count, so the same key always lands on the same partition as long as the partition count doesn't change. Without a key, the sticky partitioner batches a run of records onto one partition at a time before switching, rather than pure round robin, so full batches build up faster (this behavior changed in KIP-480, Kafka 2.4).

group.id ties multiple consumer instances together into one logical reader. Kafka assigns each partition of a subscribed topic to exactly one consumer inside that group at a time, so a topic with 12 partitions and 4 consumers in a group gives each consumer 3 partitions.

A different consumer group reading the same topic gets its own full copy of every partition. Groups don't share work with each other, only the consumers inside one group do.

In an internal, compacted topic called __consumer_offsets. Older versions stored them in ZooKeeper; that moved into Kafka itself because ZooKeeper wasn't built for the write volume a busy consumer group generates on every commit, and keeping offsets inside Kafka meant one fewer external system that had to stay available for consumers to make progress.

Time or size-based retention (retention.ms, retention.bytes) deletes whole segments once they age out or the topic grows past a size cap, regardless of which key a message carried. Compaction (cleanup.policy=compact) instead keeps only the latest value written for each key and throws away older values for that same key, so the log shrinks over time but never loses the current state for any key that's still active.

Connect standardizes moving data in and out of Kafka without custom client code for every source and sink. It handles per-connector offset tracking, retries, schema conversion (JSON, Avro, Protobuf converters), and scaling a connector across multiple worker processes, none of which the plain producer or consumer API gives you for free.

A broker is a single Kafka server process. It listens for TCP connections from producers, consumers and other brokers, stores partition data on local disk in its log directories, and serves reads and writes for whichever partitions it currently leads. A cluster is just a group of these broker processes, each with a unique broker.id, coordinating through the controller, which today runs inside the cluster's own KRaft quorum rather than a separate ZooKeeper ensemble.

When people say "add a broker," they mean starting a new JVM process with its own id and data directory, then reassigning some partitions onto it so it actually carries load. A three-node cluster with replication factor 3 means every partition's data lives on all three brokers, but only one broker at a time acts as leader for any given partition.

Each partition on disk is a directory of segment files: an active segment currently being written to, plus a set of older, closed segments. Kafka rolls to a new segment when the current one hits log.segment.bytes (1 GB by default) or log.roll.ms passes, whichever comes first.

Splitting into segments matters for two practical reasons. Retention and compaction operate at the segment level, not the message level, so Kafka can delete or compact an entire closed segment in one cheap file-system operation instead of scanning and rewriting a giant file. It also means the broker only needs an index, the .index and .timeindex files, for locating offsets within the currently relevant segments, keeping lookups fast without loading the whole log into memory.

Replication factor is how many copies of each partition Kafka keeps across different brokers: one leader and the rest followers that continuously fetch from the leader to stay in sync. A replication factor of 3 means you can lose two brokers, or take two down for maintenance, and still have every partition available with no data loss, as long as those replicas were in the ISR when the failure hit.

Replication factor of 1 means any single broker failure loses that partition's data outright, which is fine for scratch or dev topics but not for anything you care about. Most production Kafka clusters default to 3 for any topic that matters, paired with min.insync.replicas=2 so writes still require real durability, not just an acknowledgment from a single copy.

JSON messages carry no built-in contract. A producer can rename a field or change its type and every downstream consumer silently breaks or misparses data, often not discovered until someone notices bad numbers in a dashboard days later. Avro and Protobuf are binary formats where the schema is registered centrally, and each message on the wire carries a small schema id instead of the full schema text, so payloads stay compact.

On write, the registry can enforce compatibility rules, backward, forward, or full, so a producer literally cannot publish a schema change that would break existing consumers. That turns what used to be a runtime incident into a rejected write at publish time. JSON still shows up plenty in smaller systems where the operational overhead of running a registry isn't worth it yet.

A Kafka producer doesn't send one network request per message, it groups records destined for the same partition into a batch and ships the batch as one request. batch.size caps how large that batch can grow, in bytes, before it's sent. linger.ms is how long the producer will wait for a batch to fill up before sending it anyway, even if it's not full.

Set linger.ms to 0 and you get the lowest possible per-message latency but worse throughput and CPU efficiency, since you're sending lots of small requests. Bump it to 10-20ms and the producer accumulates more records per batch, which usually cuts broker-side overhead and improves compression ratio, at the cost of a small, predictable delay before messages leave the producer.

linger.ms=15
batch.size=32768
compression.type=lz4

Kafka doesn't delete individual messages as they age past retention.ms or retention.bytes, whichever limit is hit first. It deletes whole log segments once every message in that segment is older than the retention window, which is why data can survive slightly past the configured retention time, up to a full segment's worth of extra lifetime. Deletion runs on a background thread, log.retention.check.interval.ms, five minutes by default, that scans for eligible segments and removes the files from disk.

If a consumer's committed offset points into a segment that's already been deleted, that consumer gets an OffsetOutOfRangeException the next time it tries to fetch from that offset, and depending on auto.offset.reset it either jumps to the earliest available offset or the latest, silently skipping data either way.

Compression happens per-batch on the producer, and the compressed batch travels all the way to the broker's disk and back out to consumers without being re-expanded in between, which is one of Kafka's more underrated efficiency wins. gzip compresses the best but is the slowest, useful when network or disk is the bottleneck and CPU is cheap. snappy and lz4 are much faster with a lower compression ratio, a reasonable default when you want compression without adding noticeable producer latency.

zstd, the newest option, generally beats gzip's ratio while staying close to lz4 on speed, which is why it's become the default recommendation for most new clusters. The real-world impact is bigger than people expect: switching an uncompressed topic to lz4 or zstd routinely cuts network and disk usage by 60-80% for typical JSON or Avro payloads.

poll() does more than fetch records. On the first call after a consumer joins a group, it triggers the group join and partition assignment protocol if that hasn't happened yet. On every call, it sends a heartbeat to the group coordinator, checks whether any assigned partitions changed, and pulls the next batch of records from its internal fetch buffer, which is populated by background fetch requests the client issues ahead of time, so poll() usually returns immediately from memory rather than blocking on the network.

This is also why a consumer that takes too long processing the records from one poll() before calling poll() again, longer than max.poll.interval.ms, gets kicked out of the group even though the process is still alive and healthy from an infrastructure point of view.

Every partition has exactly one broker acting as its leader at any given time, and all producer writes and, by default, all consumer reads for that partition go through that one broker. The other replicas are followers that just fetch from the leader to stay caught up.

A partition briefly has no leader during a leader election, for example right after the current leader crashes or is restarted, while the controller picks a replacement from the ISR and the new leader announces itself to the cluster. During that gap, usually a few hundred milliseconds under normal conditions, producers and consumers targeting that partition get a NotLeaderForPartitionException and retry once metadata refreshes. If it drags on longer, it's usually a sign the controller itself is under stress or there's a metadata propagation problem across the cluster.

Parallelism, mostly. Multiple consumers can read different partitions of the same topic at the same time, and each partition can live on a different broker, so a topic's total throughput isn't capped by one machine's disk and network interface.

The trade-off is ordering. Kafka only guarantees order within a single partition, never across an entire topic. More partitions buys you more parallel throughput and costs you any notion of topic-wide sequence.

Ordering. Every message sharing a key lands on the same partition, and a single partition is the only place Kafka's ordering guarantee actually holds. Key by user_id and every event for one user processes in the order it happened, even though the topic as a whole has 24 partitions and no order across it at all.

A consumer joining, a consumer leaving (a crash, a deploy, or exceeding max.poll.interval.ms), or a change in the topic's partition count. Under the older eager rebalance protocol, every consumer in the group stops consuming, gives up all its partitions, and the group coordinator reassigns everything from scratch, a brief stop-the-world pause for the whole group, not just the consumer that changed.

Cooperative sticky rebalancing (KIP-429, available from Kafka 2.4 and the default assignor since Kafka 3.0) only reassigns the partitions that actually need to move. Consumers that keep their existing partitions never stop processing them, which is why most current clusters barely notice a single consumer restarting.

enable.auto.commit=true commits the latest returned offset on a timer, regardless of whether your code actually finished processing those records. Crash between that commit and finishing the work, and those records are gone from the consumer's point of view, silently. That's the at-most-once failure mode.

Manual commits (commitSync or commitAsync) called only after processing finishes shift the failure mode to at-least-once instead: crash before the commit fires and you reprocess the same batch next time, which is only safe if your downstream handling can tolerate seeing the same record twice.

Every partition has one leader replica and some number of followers. The ISR is the leader plus whichever followers have replicated recently enough to count as caught up, governed by replica.lag.time.max.ms. Fall behind that window, a slow disk, a network blip, a long GC pause, and the leader drops that follower from the ISR until it catches back up. It doesn't get removed from the cluster, just from the set eligible to become leader or count toward an acknowledged write.

acks=all tells the producer to wait for every replica currently in the ISR to acknowledge the write, not just the leader. min.insync.replicas sets a floor on how big that ISR has to be for the write to succeed at all. min.insync.replicas=2 on a replication-factor-3 topic means a write only succeeds if at least 2 replicas (the leader plus one follower) are in sync; shrink the ISR down to just the leader, and the producer gets NotEnoughReplicasException instead of a silent single-copy write.

At-most-once: a message might get lost but never processed twice, because the offset commits before processing finishes. At-least-once: a message might get processed twice but never silently dropped, because the offset commits only after processing finishes and a crash means reprocessing on restart. Exactly-once, within a Kafka-to-Kafka pipeline, combines an idempotent producer that stops duplicate writes caused by retries with transactions that make a consume-transform-produce cycle atomic across multiple partitions.

The broker assigns each producer a unique producer ID (PID), and every message gets a sequence number scoped to that PID and partition. If a network blip makes the producer retry a send that actually succeeded the first time, the broker sees a sequence number it's already recorded and discards the duplicate instead of appending it again. Enabling idempotence also forces a specific safe combination of the other producer configs.

enable.idempotence=true
acks=all
retries=2147483647
max.in.flight.requests.per.connection=5

Anywhere the topic represents current state rather than a stream of things that happened, a changelog rather than an event log. Kafka's own __consumer_offsets topic is compacted, since you only care about the latest committed offset per group and partition, not every commit ever made. Kafka Streams state store changelog topics are compacted for the same reason. A user-profile-updated topic where you only care about someone's current address is a compaction candidate; an order-placed event stream, where every individual order matters on its own, is not.

Standalone mode runs one worker process and stores offsets in a local file, fine for a laptop or a single-purpose one-off job, but if that process dies, nothing picks up where it left off automatically. Distributed mode runs multiple worker processes that form a group, much like a consumer group, and stores offsets, configs, and status in internal Kafka topics, so connectors and their tasks rebalance across whatever workers are alive the same way partitions rebalance across consumers.

kafka-topics.sh --create --topic orders --bootstrap-server localhost:9092 
  --partitions 24 --replication-factor 3 --config min.insync.replicas=2

kafka-consumer-groups.sh --bootstrap-server localhost:9092 
  --describe --group order-processing-group

Streams is a client library you embed directly in your own application, not a separate cluster you deploy and manage. There's no "Streams broker." You write a topology (map, filter, join, aggregate) against the DSL, and stateful operations like aggregations keep their working state in a local RocksDB store on each instance, backed by a compacted changelog topic so state can be rebuilt if that instance restarts somewhere else. Connect moves data between systems; Streams transforms data that's already flowing through Kafka.

StreamsBuilder builder = new StreamsBuilder();
builder.stream("orders")
    .filter((key, order) -> order.getAmount() > 0)
    .groupByKey()
    .count(Materialized.as("order-counts-store"));

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

No, only within a single partition. A topic with multiple partitions has no defined order between messages that land on different partitions, only within each partition's own append-only log. Teams that need strict global order for a topic either accept a single partition, and the throughput ceiling that comes with it, or redesign so ordering only needs to hold within a key, which is the far more common answer in practice.

session.timeout.ms governs the heartbeat channel. If the coordinator doesn't get a heartbeat from a consumer within that window, it assumes the process died and kicks it out of the group. max.poll.interval.ms is a separate check on how long you're allowed to go between calls to poll() itself, it exists because heartbeats in modern clients are sent on a background thread, so a consumer whose main thread is stuck in a slow processing loop can still look alive on the heartbeat channel while never actually calling poll() again.

Set session.timeout.ms too low and you get spurious rebalances from GC pauses or brief network blips that would have recovered fine on their own. Set max.poll.interval.ms too low relative to your actual per-batch processing time and a consumer doing legitimate slow work, say writing a big batch to a database, gets evicted mid-processing, which can cause duplicate work once it rejoins. The fix is almost never raising the timeout blindly, it's usually reducing max.poll.records so each poll() cycle finishes inside the window.

Without it, every time a consumer process restarts, say during a routine deploy, it gets a brand-new member id from the coordinator, which treats that as a new consumer joining the group and triggers a full rebalance, even though the old and new instance are really the same logical worker just restarting. Setting group.instance.id to a fixed string per instance tells the coordinator this is the same member as before, so a restart within session.timeout.ms just resumes with the same partition assignment, no rebalance, no stop-the-world pause for the rest of the group.

This matters a lot for rolling deploys of consumer fleets. Without static membership, deploying 20 consumer instances one at a time can trigger 20 separate rebalances, each one briefly pausing partition processing for the whole group.

Eager rebalancing, the original protocol, revokes every partition from every consumer in the group at the start of a rebalance, then reassigns all of them from scratch once membership settles. During that window, no consumer in the group is processing anything, which on a group with hundreds of consumers and thousands of partitions can mean several seconds of total processing pause.

Cooperative sticky rebalancing only revokes the specific partitions that actually need to move to a different consumer, everything else keeps processing uninterrupted, and it reassigns in incremental rounds rather than one big stop-the-world step. The tradeoff is that cooperative rebalancing can take more rounds of communication with the coordinator to converge, so a single rebalance can take longer wall-clock time even though the actual processing disruption is much smaller. For any group with meaningful partition counts, it's worth confirming cooperative sticky, the modern default, is actually in use, since some older configs still hard-code the range or round-robin eager assignors.

The high watermark is the offset up to which a message has been written to all replicas currently in the ISR, not just the leader. The leader tracks each follower's fetch progress and advances the high watermark only when every in-sync replica has confirmed it has that data.

Consumers, and any producer using acks=all, are only allowed to read or acknowledge up to the high watermark, never past it, because anything written to the leader but not yet replicated could disappear if the leader crashes before followers catch up. This is the mechanism that makes acks=all writes durable in practice: the producer's request doesn't get an ack until the high watermark has actually advanced past that write's offset, meaning it's confirmed on min.insync.replicas copies, not just sitting in the leader's local log waiting to be replicated.

Before leader epochs, followers used a simpler truncation rule after a leader failover: truncate your local log to whatever the new leader says its high watermark is. That worked most of the time but had a known failure mode where a follower could end up with data the new leader doesn't have, or vice versa, in certain crash-during-election sequences, leading to silent log divergence between replicas that's very hard to detect until you diff the actual bytes on disk.

A leader epoch is a monotonically increasing number stamped on every record batch, incremented every time a new leader is elected for a partition. When a follower reconnects to a possibly new leader, it first asks what's the last offset for the epoch it last saw, and the leader answers using its epoch history rather than just the current high watermark, so the follower truncates to exactly the right point instead of guessing. It's a subtle fix, but it closed a real class of replication bugs that predates it in the protocol.

Idempotence, enable.idempotence=true, which is the default now, solves duplicate writes from producer retries within a single partition, the broker tracks a sequence number per producer id per partition and drops a retried write it's already seen. That's scoped to retries, not to any notion of a business transaction.

The transactional producer is a separate layer built on top of idempotence that lets you write to multiple partitions, possibly across multiple topics, as one atomic unit, plus optionally commit consumer offsets as part of that same atomic write, the classic read-transform-write-commit pattern in Kafka Streams and similar apps. It works through a transaction coordinator broker that writes markers into each involved partition's log, and consumers configured with isolation.level=read_committed skip over any records belonging to a transaction that never committed. The cost is real: transactional writes have measurably higher latency because of the two-phase commit-style protocol and coordinator round trips, so you only reach for it when you actually need cross-partition atomicity, not for every producer by default.

Hand-rolling replication by having your application produce to two clusters means your application now owns failure handling for two independent write paths, and if one cluster is down or slow, you're either blocking your main pipeline or building your own buffering and retry logic to paper over it. MirrorMaker 2 runs as Kafka Connect source connectors that read from a source cluster and replicate to a target cluster, entirely outside your application, preserving partition structure, topic configs, and consumer group offsets, translating offsets between clusters via an offset-sync topic so failover doesn't mean every consumer starts back at the beginning.

It also handles topic renaming conventions, prefixing replicated topics with the source cluster's alias by default, so you can run active-active replication between two clusters without naming collisions. The tradeoff is replication lag, which is asynchronous and can grow under load or network issues between regions, so anything relying on MirrorMaker 2 for disaster recovery needs monitoring on replication lag specifically, not just whether the connector is running.

Before tiered storage, every byte of retained data lived on the broker's local disks, so keeping data around for 30 or 90 days meant provisioning brokers with enormous, expensive local storage just to hold cold data that's rarely read, and growing storage meant adding whole brokers rather than just adding cheaper capacity. Tiered storage lets a broker offload older log segments to cheaper object storage, S3 or equivalent, while keeping only recent, hot segments on local disk.

Reads for recent data are unaffected. Reads that fall back into the archived range get fetched from remote storage transparently, with higher latency but without needing to keep everything locally. In practice this decouples retention length from local disk sizing, so you can offer months of retention on a topic without running brokers with terabytes of local NVMe just to satisfy a compliance requirement that rarely gets read.

A Single Message Transform is a small, chainable transformation applied to each record as it passes through a connector, before it hits a sink or after it leaves a source, things like renaming a field, masking a sensitive value, dropping a tombstone, or casting a type. They're configured declaratively in the connector's JSON config, no custom code needed for the common cases, which keeps simple transformations out of application code entirely.

A dead-letter queue configuration on a sink connector redirects records that fail to process, bad JSON, schema mismatch, whatever the sink's converter chokes on, to a separate topic instead of stopping the whole connector task, which would otherwise block every other record behind that one poison message. The real judgment call is what counts as recoverable enough to skip and move on versus a signal the connector should stop because continuing means silently losing data downstream. For something like a payments pipeline you generally want the connector to stop and page someone rather than quietly routing malformed transaction records to a DLQ and moving on.

Setting broker.rack tells Kafka which physical failure domain each broker lives in, an actual server rack, an availability zone, whatever boundary represents things that can fail together. With rack awareness configured, Kafka's replica placement algorithm spreads a partition's replicas across different racks rather than just different brokers, so a replication factor of 3 across three racks actually protects you if an entire rack or AZ goes down, not just if one individual broker process crashes.

Without it, it's entirely possible, and this does happen in practice, for all three replicas of a partition to land in the same AZ purely by chance of broker id ordering, which means an AZ outage takes that partition fully offline even though replication factor was correctly set to 3. It's a cheap config to set on cluster provisioning and expensive to fix later, since fixing placement after the fact means reassigning partitions across the cluster, a real I/O-heavy operation on a live cluster.

Quotas cap how much throughput, bytes per second produced or consumed, or how much request handler time a given client id or user can consume, configurable per-client or per-user via dynamic broker configs. When a client exceeds its quota, the broker doesn't reject the request, it throttles it, delaying the response long enough to bring the client's rate back within quota, which the client sees as increased request latency plus a throttle-time metric if it's watching for it.

This matters most in multi-tenant clusters, where one noisy team's badly-tuned producer, say sending uncompressed, unbatched, tiny messages, can otherwise eat enough broker I/O capacity to degrade latency for every other topic on the cluster. Setting sensible default quotas per client id, then raising them deliberately for teams that need more, is far less painful than discovering the problem during an incident review.

Just changing the schema and deploying only works safely if the change is compatible under whatever compatibility mode the Schema Registry enforces for that subject, and removing a required field without a default is backward-incompatible by definition, since old consumers reading new messages get a missing field they expect to exist. The registry actually rejects that schema registration outright if compatibility mode is set to BACKWARD or FULL, which is usually a good thing, it forces the conversation before it becomes an incident.

The real pattern for a genuinely breaking change is a multi-step rollout: first mark the field deprecated but keep producing it, get every consuming team to confirm they've removed their dependency on it, only then actually drop it from the schema. For a truly incompatible restructure, standing up a new topic version, topic-name-v2, and running both in parallel while consumers migrate is usually less painful than forcing a hard cutover on a shared topic other teams depend on and you don't fully control the deploy timing of.

num.network.threads controls how many threads handle reading requests off the network and writing responses back, the threads doing socket I/O and initial request parsing. num.io.threads controls how many threads actually process those requests once parsed, the ones doing the real work like appending to the log or reading from it. If network threads are maxed out, requests queue up before they're even fully read off the wire, which shows up as elevated request latency with the broker's disk and CPU looking otherwise idle, a classic sign of network-thread starvation rather than genuine overload.

If IO threads are the bottleneck instead, you'll typically see low request-handler-avg-idle-percent alongside actual disk queueing. In practice you rarely need to touch these until you're running brokers with a lot of cores and high connection counts, and the fix is almost always increasing num.io.threads to roughly match core count for a disk-heavy workload, not touching network threads, which most workloads don't come close to saturating.

KRaft (from KIP-500) replaced the external ZooKeeper ensemble with a small set of Kafka nodes running a Raft-based controller quorum that stores cluster metadata inside Kafka itself instead of a separate system. It reached production-ready status in Kafka 3.3 in October 2022, and later major releases dropped ZooKeeper support entirely.

I don't have a clean number on how many production clusters have actually finished migrating off ZooKeeper as of mid-2026. Plenty of older clusters are probably still running it. If you're studying today, learn KRaft first, but don't be surprised if the cluster you inherit at a new job still has a ZooKeeper ensemble bolted on.

The hash-to-partition mapping shifts for most keys the instant partition count changes, since it's a mod against the new count. A key that used to land on partition 3 might now land on partition 17. Anything downstream relying on "everything for this key arrives on the same partition, in order" loses that guarantee right at the changeover, old messages for a key sit on the old partition, new ones for the same key land somewhere else.

That's the real reason teams that expect to grow tend to over-provision partitions from day one instead of resizing a live keyed topic later. Going from 12 to 24 sounds harmless in a design doc. It's a genuinely disruptive operation once real keyed traffic depends on the old mapping.

max.poll.interval.ms is probably too short for how long the consumer really takes to process a batch. If processing what poll() returned runs longer than that interval, Kafka assumes the consumer is dead, kicks it out of the group, and triggers a rebalance, even though the process is still alive and working through the batch.

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    consumer.commitSync(); // only after the whole batch is processed
}

Fix: raise max.poll.interval.ms, shrink max.poll.records so each batch takes less wall-clock time, or move the slow part (a downstream API call, a heavy transform) off the polling thread entirely so poll() keeps getting called on schedule.

Normally Kafka only elects a new leader from a replica that was in the ISR, which guarantees no committed data gets lost in the handoff. Unclean leader election allows an out-of-sync replica to become leader anyway when no in-sync replica is available, trading data loss for keeping the partition writable at all. unclean.leader.election.enable defaults to false, and I'd leave it there for almost any workload I can think of.

I've heard the case made for flipping it on for latency-sensitive metrics pipelines, where a small gap of dropped data beats a stalled partition. I've never actually run that configuration in production myself, so take that specific argument secondhand.

It only covers Kafka-to-Kafka. If your consumer reads a message and then calls an external payment API, sends an email, or writes to a database outside a Kafka transaction, none of that is covered. You can still fire that side effect twice if the crash happens after the side effect but before the offset commit lands.

My honest take: exactly-once gets marketed like it solves the general "don't duplicate side effects" problem, and it solves a narrower one than that. If your sink isn't Kafka, or something transaction-aware like a Kafka Streams state store, you still have to make that sink idempotent yourself, an upsert keyed by message id, a unique constraint, something along those lines.

processing.guarantee=exactly_once_v2 wraps the read-process-write cycle in a Kafka transaction, so the consumed input offset commit and the produced output records commit or roll back together. I don't think most teams need it by default. It adds real overhead, transaction coordinator round trips, longer commit latency, for a guarantee that only matters if a duplicate downstream write is genuinely expensive to undo. A lot of Streams apps I've seen turn it on reflexively and never measure whether at-least-once plus an idempotent output sink would've covered the actual requirement.

Lag that creeps up slowly under normal load usually means the consumer group is under-provisioned for current traffic, not that something is outright broken, and a static "alert if lag exceeds 1,000" threshold either fires constantly during a legitimate spike or stays quiet while lag climbs for hours during a slow leak. Track lag as a rate relative to incoming message rate instead of a fixed absolute number, and separately watch for lag spikes that line up exactly with rebalance events, since a rebalance briefly pausing every consumer in a group is one of the most common causes of a lag graph that looks like an outage but isn't one.

This is the classic split-brain concern: broker A is leading partition P, and it loses connectivity specifically to the controller, and possibly the KRaft quorum, while still being reachable by producers, consumers, and its own followers on a different network path. The controller, seeing broker A's session expire, declares it dead and elects a new leader from the remaining ISR, say broker B, and B starts accepting writes. Meanwhile A doesn't necessarily know it's been deposed yet, since nothing has told it directly, it's only isolated from the controller, not from clients.

The protection is that every broker, on every request it handles, includes a controller epoch and leader epoch that clients and followers check against what they last saw from the controller's metadata broadcasts. Once A's connection to the controller resumes, or once clients refresh metadata, which happens on a timer and on certain errors, they discover B is now the leader with a higher leader epoch, and A's requests start getting rejected with a stale-leader error, forcing it to step down. The window where a zombie old leader can still serve stale reads or accept writes that will later be truncated is real but bounded by how fast metadata propagates, typically the metadata refresh interval plus however long the network partition actually lasts. This is exactly why acks=all with min.insync.replicas=2 matters here: a write accepted only by the zombie leader, with no follower confirming it, never gets acknowledged to the producer in the first place, so the producer knows to retry against whatever broker metadata resolves to next.

When a transactional producer begins a transaction, the coordinator assigns it a producer id and epoch and tracks which partitions it has written to during that transaction. Every record written during the transaction is tagged with that producer id and epoch, but the record itself isn't visible to consumers yet, it just sits in the log like a normal record. When the producer commits, the coordinator writes a special control record, a commit marker, into every partition the transaction touched, and only after that marker is durably written does the transaction officially count as committed.

A consumer configured with isolation.level=read_committed doesn't just read records in offset order and hand them to your application. It buffers records belonging to open transactions and only releases them once it sees the corresponding commit marker for that producer id and epoch land in the log. If it instead sees an abort marker, from an explicit abort or the coordinator timing out a producer that went silent, it discards the buffered records for that transaction entirely, they never reach your application code. This is why read_committed consumers see slightly higher latency, they're intentionally holding back data until the transaction resolves, and why a producer that crashes mid-transaction and never comes back eventually has its transaction aborted by the coordinator after transaction.timeout.ms, cleaning up the dangling writes rather than leaving them stuck in limbo forever.

Picture three replicas: leader A, followers B and C. A accepts a write at offset 100 and immediately crashes before any follower has fetched it, so the high watermark, which only advances once followers confirm, is still sitting at 99. B gets elected the new leader, it was in the ISR, just slightly behind on that last write. B accepts a new, different write at offset 100, a completely different record than what A had locally. Now A restarts and rejoins as a follower.

Using only the high watermark rule, A would truncate to the new leader's high watermark, but the old protocol before leader epochs had cases where the truncation calculation used stale information about what the follower itself had confirmed, and under certain crash-during-election timings could leave A retaining a stale version of its own offset-100 record instead of adopting B's version, quietly diverging the two logs at that offset while both brokers believe they're in sync. Leader epochs close this by making truncation decisions epoch-aware rather than just watermark-aware. Every time leadership changes, the epoch increments, and the leader keeps a small durable log of which epoch started at which offset. When A rejoins, instead of asking what's your high watermark, it asks what's the last offset for epoch N for the last epoch it actually saw locally, and B answers using its epoch history, which correctly reflects that A's epoch ended before offset 100 was ever confirmed under that epoch. A is told to truncate its log back to exactly where its old epoch ended, discarding its version of offset 100 entirely and re-fetching B's version instead. It replaces an ambiguous watermark comparison with an unambiguous, authoritative epoch-history comparison.

Kafka Streams backs every stateful operation, aggregations, joins, windowed counts, with a local RocksDB store on the instance's disk, and every write to that store is also published to an internal changelog topic, a compacted or windowed-and-compacted Kafka topic that mirrors the store's contents. If the specific broker holding the leader replica for that changelog partition goes down, normal Kafka replication kicks in first, assuming replication factor is at least 2 for internal topics, which it should be, a common misconfiguration people hit only when it's too late, a follower gets promoted to leader and the changelog topic itself doesn't lose data.

The recovery scenario that actually matters is when the Streams instance itself, the one holding the local RocksDB store, is lost or restarted, not the broker. In that case, when the task gets reassigned, to the same instance restarting or a different one entirely, Streams has to rebuild the RocksDB store from scratch by replaying the entire changelog topic from the beginning, which for a large, long-lived aggregation store can take real wall-clock time, minutes, sometimes longer, during which that partition's processing is stalled. This is exactly what standby replicas, num.standby.replicas, are for: a standby instance keeps a warm, continuously-updated copy of the store on a different node by consuming the changelog in the background, so when the active task needs to move, it can fail over to the standby's already-warm store instead of rebuilding from offset zero. Skipping standby replicas is a common cost-cutting decision that looks fine until the first real failover, when a stateful topology partition goes dark for several minutes instead of failing over near-instantly.

This pattern almost always points away from raw resource exhaustion and toward something structural: request queueing, GC pauses, or a specific slow partition dragging down an otherwise healthy broker's aggregate numbers. First I'd pull request-handler-avg-idle-percent and network-processor-avg-idle-percent per broker, not cluster-wide, because a cluster average can look completely healthy while one specific broker is pegged, and if that broker happens to be leading a disproportionate share of high-traffic partitions, worth checking directly, it drags down latency for anyone touching those partitions specifically.

Second, I'd check GC logs on the suspect broker. Kafka brokers are JVM processes, and a long stop-the-world pause, even a couple hundred milliseconds, shows up exactly as a latency spike with otherwise-idle-looking CPU and disk graphs, since the broker simply isn't doing anything during the pause, including not using CPU. Third, I'd look at request queue time specifically, there's a metric for time spent in the request queue versus actual local processing time versus remote or replication time, broken down per request type, because produce requests waiting in queue versus produce requests slow to actually replicate point at completely different fixes: the first is a thread pool sizing problem, the second might be a slow follower or a network issue between specific brokers. And fourth, if it's specifically acks=all producers seeing the timeouts, I'd check whether a particular follower is lagging, under-replicated partition count, replica fetch lag, since with min.insync.replicas enforced, the entire write path is only as fast as the slowest in-sync replica has to confirm, one slow disk on one follower broker can bottleneck every acks=all write for every partition it follows.