Apache Kafka Scenarios & Design

The reasoning

**Shape first.** `kafka-consumer-groups.sh --bootstrap-server broker:9092 --describe --group orders-writer` gives the committed offset, log-end offset and lag for every partition, and which consumer owns it. Check the input side too: if producers suddenly doubled their rate — a backfill, an upstream retry storm — the consumers may be fine and simply outrun. Lag is arrival rate minus service rate, and either side can move.

**Uneven lag** points at one partition. A hot key shows as one partition with far more incoming bytes than the rest. A poison record shows as a committed offset that is not moving at all, with the consumer restarting in its logs. A sick instance — a bad node, long garbage-collection pauses — shows as every partition owned by that one consumer lagging together. A frozen offset and a slowly moving one are different problems.

**Even lag with idle CPU** means the consumers are blocked. The usual causes are a slow sink (look at the p99 of the write, not the average), a synchronous call per record, `commitSync()` after every record, or rebalance churn — every rebalance pauses consumption, so check the logs and the group's rebalance rate. The client metrics `records-lag-max`, `fetch-latency-avg` and `poll-idle-ratio-avg` tell you whether the consumer is waiting on Kafka or on its own processing.

The fixes follow the cause: batch and parallelise sink writes (keyed so per-order ordering holds), stop committing per record, stabilise the group with static membership or the KIP-848 protocol, fix the key. Only if all 12 consumers are genuinely busy is more parallelism the answer — and in a classic group that means more partitions. Meanwhile, work out the headroom: how many hours until the oldest unread record falls out of retention.

The reasoning

More partitions cannot help. The partition is `hash(tenant_id) % N`, and one key always lands on one partition whatever N is. Going from 24 to 48 moves the hot tenant; it does not split it — and it remaps every other tenant too.

The options, from least to most disruptive: **salt the known hot keys** (key = `tenant_id#bucket`, bucket = `hash(session_id) % 8`, applied only to tenants on a configured list); **key everyone by a finer entity** such as session or user, if nobody needs tenant-wide order; a **custom partitioner** that gives large tenants dedicated partitions; or a **separate topic** for the whale with its own consumer pool.

Every option shrinks the ordering scope for the affected tenant, from tenant to bucket or session. So say what needs ordering: for a clickstream it is usually the session, not the tenant. Consumers that aggregate per tenant need a two-stage aggregation — partial results per bucket, then combined — and during the switch-over old records keep their old keys, so consumers must tolerate both for a while.

Detect skew before a customer does: incoming bytes per partition on the brokers, lag per partition on the consumer side, and a sample of key frequencies at the producer. A null key would spread perfectly evenly and give up ordering entirely.

The reasoning

The usual arithmetic is partitions ≥ max(T / P, T / C), where T is target throughput, P what one partition can take from producers and C what one consumer can process from a partition. Consumers usually bind: 400 / 10 = 40. Measure P and C yourself — both depend on record size, compression, acks and what the consumer does per record — rather than borrowing someone else's number.

Size for the future rather than today because adding partitions to a keyed topic later remaps keys (`hash % N` changes), which breaks per-key ordering for anything still in flight. Prefer a count with many divisors — 48, 60 — so that 6, 8, 12 or 16 consumers divide it evenly.

The costs of too many: open segment and index files on every replica, more replica-fetcher work, more metadata for the controller, more leaders to move when a broker fails, more assignment work on every rebalance, and producer batches spread thinner across partitions, which means smaller batches and less throughput per request. And you can never go back down.

Partition count is not the whole capacity plan. 400 MB/s written with replication factor 3 is 1.2 GB/s of disk writes across the cluster before any consumer reads a byte; broker count, disks and network are sized from that.

The reasoning

Mechanically, partitions can only be added. Existing segments are untouched and the new partitions start empty. Producers notice on their next metadata refresh (`metadata.max.age.ms`, five minutes by default) and start hashing modulo 48; consumers rebalance to pick up the new partitions. One gotcha: with `auto.offset.reset=latest`, records written to a new partition before any consumer has an offset for it can be skipped, which is one reason to use `earliest` on topics that may grow.

Ordering is what breaks. Customer 42 lived on partition 7, which still holds two million unprocessed events for it. After the change its new events go to, say, partition 31, which is empty and gets processed first. For a consumer maintaining per-customer state — balances, order status — that is corruption, not a delay. Adding partitions to a keyed topic is only safe when producers are paused and consumers have fully drained, which is exactly what you cannot do while lagging.

The alternatives. **Key-ordered parallelism inside each partition**: one consumer per partition fans records out to worker lanes by `hash(customer_id)`, so different customers run concurrently and each customer stays in order; open-source libraries such as Confluent's Parallel Consumer do this. **A new topic** with 48 partitions, producers cut over at a known point and consumers switching once the old topic drains. **Share groups** (production-ready in Kafka 4.2), if records are independent. And first, **find out why each record is slow** — the answer is often the sink.

Whatever you do, the partition count can never come back down.

The reasoning

With `linger.ms=0` the sender ships whatever is waiting the moment it is free, so under moderate load batches are small, and a 16 KB batch holds only 16 of these records even when full. Per-request overhead dominates. (Kafka 4.0 changed the `linger.ms` default from 0 to 5 for exactly this reason, so a team at 0 has set it explicitly or runs an older client.)

The settings: `linger.ms` trades a few milliseconds of latency for fuller batches; `batch.size` is per partition, so a producer writing to many partitions needs `buffer.memory` large enough to hold all the open batches; `compression.type` (none by default; gzip, snappy, lz4 or zstd) compresses each batch in the producer, the broker stores it compressed, and repetitive JSON usually shrinks substantially — measure it on your own data. `max.in.flight.requests.per.connection=5` with idempotence keeps the pipe full without reordering.

Look for the non-config culprits too. The commonest is calling `send(record).get()` on every record, which makes the producer wait for a round trip per record. After that: every record carrying the same key, so one partition takes everything; expensive serialisation; and the network between producer and brokers.

Measure rather than guess: the producer metrics `batch-size-avg`, `records-per-request-avg`, `record-queue-time-avg` and `compression-rate-avg`, and `kafka-producer-perf-test.sh` against the candidate settings. Do not promise a number before you have run it.

The reasoning

The timeline: at t0 the consumer polls 500 records. At t0 + 5 minutes `max.poll.interval.ms` expires, the client leaves the group and a rebalance gives its partitions to another member, which starts from the last committed offset — before this batch. At t0 + 8 minutes the original finishes; its writes are already in the warehouse, and its commit fails with `CommitFailedException`. The same records are now being written again. With every member doing this, the group rebalances continuously, and under the eager protocol every rebalance pauses everyone.

Fix the arithmetic first: 500 records at about a second each is eight minutes. Lower `max.poll.records` to around 50 so a batch takes under a minute, and set `max.poll.interval.ms` above the real worst case — not far above it, because it is also how long a genuinely hung consumer goes unnoticed.

Then fix the design. Call the API concurrently with bounded parallelism so a batch takes seconds, cache enrichment results that repeat, or `pause()` the partitions, process on a worker pool, keep calling `poll()` to stay in the group, and `resume()` when the batch is done.

Duplicates will still happen — deploys and scaling cause rebalances too. Commit in `onPartitionsRevoked`, move to cooperative or KIP-848 rebalancing so fewer partitions pause, use static membership for rolling restarts, and make the warehouse load a MERGE keyed on the event ID so a reprocessed batch changes nothing.

The reasoning

The mechanics: a table of `(group, topic, partition, next_offset)`; `enable.auto.commit=false`; each batch inserts its rows and updates `next_offset` in one transaction. A `ConsumerRebalanceListener` reads the stored offsets in `onPartitionsAssigned` and calls `consumer.seek(partition, offset)`. A crash mid-transaction commits nothing, and the records are read again from the stored position.

Its limits. It only works when the entire effect of a record lands in that one transactional store — add a cache write or an API call and they are outside the transaction. It cannot catch duplicates that are already in the topic, because an upstream retry that produced a record twice gave it two offsets. And standard lag monitoring reads Kafka's committed offsets, so you lose it unless you also commit to Kafka asynchronously, purely for observability.

The processed-events approach keeps normal Kafka commits (at-least-once) and makes the write refuse a repeat: `INSERT ... ON CONFLICT (event_id) DO NOTHING` in the same transaction as the business effect, or an upsert on the business key that only applies a higher version. It tolerates any source of replay. The cost is a unique index and keeping dedup keys for at least the longest replay window — partition the table by day and drop old partitions.

For payments specifically, the charge itself lives at the payment provider, outside any transaction you own. Pass the event ID as the provider's idempotency key, so a replayed event cannot charge twice even if everything on your side runs it again.

The reasoning

**Producer.** `acks=all` and `enable.idempotence=true` are the defaults since Kafka 3.0, but check nobody overrode them — setting `acks=1` quietly disables idempotence. `delivery.timeout.ms` (120000 ms) bounds retries. Then the part most teams miss: the send callback. A fire-and-forget `send()` whose exception is only logged is the source of most "Kafka lost my message" incidents; on failure the order service must store the order somewhere durable or fail the request.

**Topic and broker.** Replication factor 3 across racks or availability zones (`broker.rack` makes replica placement rack-aware), `min.insync.replicas=2` so acks=all means at least two copies, and `unclean.leader.election.enable=false` (the default) so an out-of-sync replica never becomes leader. Watch the broker defaults: `default.replication.factor=1`, `min.insync.replicas=1` and `auto.create.topics.enable=true` mean a typo in a topic name creates a single-copy topic that acks=all will happily acknowledge. Turn auto-creation off and create topics deliberately.

**What still loses data.** Kafka does not fsync each write by default (`log.flush.interval.messages` is effectively unlimited); it relies on copies on several machines. A correlated failure — all three replicas losing power before the page cache is flushed — can lose acknowledged records, which is why the replicas belong in different zones.

**Consumer.** `enable.auto.commit=false`, commit after the write succeeds, and make the write idempotent, because at-least-once delivery is what the rest of this setup buys you. Keep retention long enough to replay after a downstream incident.

The reasoning

The loop: a consumer with `enable.auto.commit=false`, a producer with a `transactional.id`, `initTransactions()` once, then for each poll `beginTransaction()`, send the fees, `sendOffsetsToTransaction(offsets, consumer.groupMetadata())`, `commitTransaction()`. On an error, `abortTransaction()` and rewind to the last committed offsets; on `ProducerFencedException`, close — another instance has taken over.

Kafka Streams wraps the same thing: `processing.guarantee=exactly_once_v2` (the default is `at_least_once`) needs brokers on 2.5 or later, and by default a cluster of at least three brokers, because the transaction state log wants three replicas with a minimum ISR of two. Under exactly-once the default `commit.interval.ms` drops from 30 seconds to 100 ms, which bounds how long read_committed readers wait for results.

What quietly breaks it: a side effect outside Kafka inside the loop — a database write, an HTTP call, an email — which is repeated when a transaction aborts and replays; a downstream consumer left on the default `read_uncommitted`, which reads records from aborted transactions; another, non-transactional producer writing to the same output topic; or logic that is not deterministic (wall-clock time, random numbers), so a replayed transaction writes different values.

The cost is transaction markers, coordinator round trips, and end-to-end latency of roughly the commit interval for read_committed readers. The alternative is at-least-once with an idempotent consumer downstream, which is often enough.

The reasoning

`kafka-consumer-groups.sh` resets a group's committed offsets to a time (`--to-datetime`), a duration back (`--by-duration PT6H`), a relative shift (`--shift-by`) or an explicit offset, with `--dry-run` to preview and `--execute` to apply. The group must have **no active members** while you do it. Under the hood it uses the time index: for each partition, the earliest offset whose timestamp is at or after the given time.

Two conditions. The 6 hours must still be inside retention, which with a seven-day default they are. And the timestamps must mean what you think: by default a record carries the producer's `CreateTime`, so producers with skewed clocks put the reset point in the wrong place; topics using `LogAppendTime` get the broker's clock instead.

The sink decides whether the replay fixes the data or doubles it. An append-only loader writes the 6 hours a second time. A MERGE on the order ID overwrites the wrong amounts; a delete-then-reload of the affected window works too. Either way, tell the consumers of the warehouse table the window is being rewritten.

If stopping ingestion is not acceptable, leave the production group alone and run a **separate backfill group** with its own `group.id`, which seeks to the start time, reads up to the end offsets captured when it started, and stops. The live group keeps its latency.

The reasoning

Deserialisation happens in the client while `poll()` builds the records, so a record that cannot be decoded fails the poll itself. Recent Java clients throw a `RecordDeserializationException` that carries the partition and offset. Without handling, the consumer crashes, restarts from the committed offset, and hits the same bytes again — forever, on that partition only.

Three ways out. Consume with a byte-array deserialiser and decode in your own `try`/`catch`, so a bad record is an ordinary value you can route. Use your framework's error-handling deserialiser — Spring Kafka's `ErrorHandlingDeserializer` with a dead-letter publishing recoverer is the common one. Or catch `RecordDeserializationException`, publish what you can, and `seek()` to the offset after it.

A dead-lettered record needs enough to be fixed and replayed: the original key and value bytes, the source topic, partition and offset, the exception, a timestamp and the consumer's version, usually in headers. Alert on any dead-letter traffic; a dead-letter topic nobody watches is data loss with extra steps.

Separate poison from transient. A record that can never be decoded goes straight to the dead-letter topic; a record whose downstream call timed out should be retried. Schema Registry serialisers stop most malformed records at the producer, and Kafka 4.2 share groups add broker-side delivery counts (5 attempts by default) — but those apply to share groups, not to a classic consumer group.

The reasoning

Classify first. A validation error (the API says the order is invalid) will never succeed and goes straight to the dead-letter topic. A timeout or a 5xx is transient and worth retrying.

The tiers: main topic, then `orders.retry.1m`, then `orders.retry.10m`, then `orders.dlq`, with headers carrying the attempt count, the original topic, partition and offset, the first failure time and the last error. A retry-topic consumer waits until the head record is due by pausing its partition rather than sleeping in the poll loop, which would trip `max.poll.interval.ms`. Records land in a retry topic in the order they failed, so the head is always the earliest due.

Ordering is the cost. If order 42's `paid` event goes to the retry topic and its `shipped` event succeeds on the main topic, the sink sees shipped before paid. Either track the keys that have a record in retry and divert their later events there too, preserving per-key order; or carry a version per order and have the sink ignore anything older than what it holds.

And notice when the problem is the whole API, not individual records. If every call fails, retry topics just move the backlog around. Open a circuit breaker, `pause()` the consumer and let lag absorb the 20 minutes — with days of retention, lag is the cheapest buffer there is, and ordering is untouched. Share groups in Kafka 4.2 offer per-record redelivery without blocking a partition, but only for work that does not need ordering.

The reasoning

The default is `errors.tolerance=none`: any error fails the task immediately, its status becomes FAILED, and it stays that way until someone restarts it through the REST API. That is correct when a skipped record is a correctness bug, and painful when nobody is alerted.

The configuration: `errors.tolerance=all`, `errors.deadletterqueue.topic.name`, `errors.deadletterqueue.context.headers.enable=true` (which adds headers with the original topic, partition, offset and exception), and `errors.log.enable=true`. The dead-letter topic is created with replication factor 3 by default. `errors.retry.timeout` covers transient failures before a record is given up on.

What it covers: failures in converters (deserialisation) and single message transforms. Failures inside the connector's own write to S3 are the connector's business; connectors that support the errant-record reporter can send those to the dead-letter topic too, others only retry or fail.

Confluent's documentation is explicit that the dead-letter queue applies only to sink connectors. On the source side a failure happens while reading the external system, before any Kafka record exists to route. And `errors.tolerance=all` without a dead-letter topic simply drops the bad records — so alert on the connector's error metrics whichever way you configure it.

The reasoning

A rename is a delete of `amount` plus an add of `total`. If `total` has a default, the change passes a BACKWARD check: a reader with the new schema can read old records and fills in the default. The producer deploys first and starts writing records without `amount`. Consumers still on the old schema expect `amount`, which has no default, cannot resolve the new records, and fail. The registry did what it was configured to do.

Confluent's documentation spells out what each mode implies. **BACKWARD** (the default, chosen because it lets consumers rewind to the start of a topic): upgrade consumers first. **FORWARD**: old readers can read new data, so upgrade producers first. **FULL**: both, in any order. The `_TRANSITIVE` variants check against every previous version, not just the latest.

A safe rename: version 2 adds `total` with a default and keeps `amount`; producers write both; consumers move to `total`; version 3 removes `amount` once nothing reads it. Avro aliases on the reader side are the other way to carry an old name across.

The guard rails: set compatibility explicitly per subject (`FULL_TRANSITIVE` for contracts many teams consume), turn off `auto.register.schemas` in production producers and register schemas from CI with a compatibility check against the registry, and give consumers a dead-letter path for records they cannot decode, so one bad deploy does not become a crash loop.

The reasoning

**Clarify first**, because that is what the reporting candidate was marked down for: message size, peak rate, retention, ordering scope, delivery guarantee, number of consumer groups, multi-region or not. At 1 million 1 KB messages a second you are writing about 1 GB/s — roughly 86 TB a day before replication — which settles that storage is on disk and that the design must lean on sequential I/O.

**Storage.** Each partition is a sequence of segment files with a sparse offset index and a time index. Writes are appends, reads are sequential, the OS page cache does the caching, and data goes from the page cache to the consumer's socket without being copied through the application (zero-copy). Ordering is per partition; records with the same key hash to the same partition, which gives per-key order.

**Replication and metadata.** Each partition has a leader and followers; the followers that keep up form the in-sync set, and a write is committed — and acknowledged under acks=all — when the whole set has it. The high watermark is how far consumers may read. A controller quorum using Raft stores the metadata as a log and elects partition leaders, with leader epochs so a deposed leader cannot keep accepting writes. This is where the reported answer discussed ZooKeeper and its replacement by KRaft.

**Consumers.** Pull, with long polling. A group coordinator per consumer group tracks members by heartbeat, assigns partitions (incrementally, to avoid stop-the-world pauses) and stores committed offsets in an internal compacted log. At-least-once is process-then-commit. Scaling out means adding brokers and moving partitions with throttled copies; there is no global order and no per-message acknowledgement, which is exactly what makes the rest cheap.

The reasoning

**Topics.** Services publish to `notifications.requested`, keyed by customer ID so one customer's notifications stay in order. A router consumes it, applies preferences and quiet hours, and writes to `notif.push`, `notif.sms` and `notif.email`. If the router writes its outputs and its input offset in one Kafka transaction, the fan-out itself is exactly-once — inside Kafka. Separate transactional notifications (order out for delivery) from promotional ones, so a marketing blast can never delay them.

**Senders.** One consumer group per channel, scaled independently to each provider's rate limits, pausing when a provider throttles, with tiered retries and a dead-letter topic.

**Deduplication.** The duplicate comes from the gap between calling the provider and committing the offset: a crash in between resends. So each sender claims `(notification_id, channel)` in a store with an atomic set-if-absent — Redis `SET NX` with a TTL, or a unique key in a database — before sending, and skips claimed ones. That makes a crash after claiming and before sending a lost notification instead of a duplicate, so choose per channel: a one-time password should err towards at-least-once, a promotion towards at-most-once. Where the provider accepts an idempotency key, pass the notification ID and let the provider close the gap.

**Scale.** Traffic peaks at lunch and dinner; size partitions per channel for the peak, and remember that SMS and email throughput is capped by the providers, not by Kafka.

The reasoning

**Size.** 500,000 events a second at around 200 bytes each is about 100 MB/s before replication. The partition count comes from what one processing task can handle — measure it — and then headroom: if a task handles 20,000 events a second, that is 25 tasks, so 48 or 64 partitions.

**Keys.** Keying by `driver_id` spreads load evenly and keeps each driver's pings in order, which you need, because "active drivers per cell" means distinct drivers whose latest position in the minute falls in that cell. So it is two stages: the latest position per driver per window, then a count per cell. The cell-keyed stage is where skew lives — an airport or a stadium is far denser than anywhere else — which is why aggregations like this pre-aggregate per task and then combine.

**Windows.** Event-time tumbling windows of one minute, with a watermark that allows a few seconds of lateness for mobile networks. Results go to Redis (latest count per cell, with a TTL) for the live map, and an hourly roll-up goes to Postgres or the lake, kept for a month. The raw topic only needs hours or days of retention.

**Guarantees.** Counting distinct drivers per window is naturally idempotent, and writing results as an upsert keyed by `(cell, minute)` means an at-least-once replay rewrites the same numbers — exactly-once is not needed here.

The reasoning

The source options that matter: `kafka.bootstrap.servers`; `subscribe` (or `subscribePattern`); `startingOffsets`, which applies only to a brand-new query and defaults to `latest` for streaming; `maxOffsetsPerTrigger` to cap how much one micro-batch reads when there is a backlog; and `failOnDataLoss`, true by default. Key and value arrive as binary and need casting, `from_json`, or an Avro decoder that understands the registry's wire format.

For each micro-batch, the driver decides the end offsets, writes them to the checkpoint's offset log **before** running the batch, and records completion in the commit log afterwards. The Delta sink records the batch ID with its transaction, so a batch that is re-run after a crash is recognised and not written twice. That combination — replayable source, checkpoint and idempotent sink — is how the pipeline gets exactly-once into Delta. Spark does not commit offsets to the Kafka consumer group, so lag tools that read committed offsets see nothing unless you publish progress separately.

On restart the query resumes from the checkpoint. If Kafka has already deleted those offsets through retention, `failOnDataLoss=true` fails the query, which is the right default; `false` skips the gap silently. Delete the checkpoint and it is a new query: `startingOffsets` applies again, so `latest` skips everything since and `earliest` re-reads the whole topic — duplicates, unless the table is loaded with a MERGE on the key.

For parallelism, Spark maps Kafka partitions to tasks one to one by default; `minPartitions` splits them further without touching the topic.

The reasoning

The chain of guarantees: the same key goes to the same partition (while the partition count is fixed); a partition is read in order by one member of the group; and then **your code must not reorder it**. A thread pool over the whole batch, with no regard for keys, throws away the ordering Kafka just gave you.

**Key-affine lanes.** Inside each consumer, N single-threaded lanes, lane = `hash(order_id) % N`, each processing first-in first-out. Throughput scales with N and every order is still serial. Because lanes finish out of offset order, the commit per partition is the highest offset below which everything is done. Open-source libraries such as Confluent's Parallel Consumer implement this key-ordered mode.

**Make the sink safe.** At-least-once delivery means replays. Have the order service stamp a per-order sequence number, and apply updates conditionally: update only if the incoming sequence is higher than the stored one. A duplicate or a straggler then does nothing.

**Check the producers.** If `paid` comes from the payments service and `packed` from the warehouse service, keying both by `order_id` puts them in one partition but not in any meaningful order — they were appended by two independent producers. The per-order sequence (or an outbox in the owning service) is then what establishes order, not Kafka. More partitions or a share group are the wrong tools here: the first breaks live ordering, the second gives it up.

The reasoning

**Topics and keys.** Debezium writes each table to its own topic and keys every event by the row's primary key, so all changes to one order are on one partition in the order Postgres committed them. The value is an envelope with `before`, `after`, `op` (create, update, delete, or read during a snapshot) and `source` metadata including the log position. A delete produces a delete event and then, by default, a tombstone: the key with a null value.

**Retention.** `cleanup.policy=compact` keeps the latest event per key, so a consumer that starts from the beginning — a cache after a flush, a rebuilt index — gets the current state of every row, not seven days of history. Tombstones are removed after `delete.retention.ms` (one day by default), so a consumer that is offline longer than that can miss a delete. `min.compaction.lag.ms` guarantees a window in which every change stays visible. `compact,delete` bounds disk as well, at the cost of no longer holding complete state.

**Consumers.** The warehouse applies a MERGE on the primary key, ordered by log position within a key, turning `op=d` into a delete or a soft-delete flag. The search index upserts or deletes the document by ID; the cache sets or evicts. All three must be idempotent, because Connect is at-least-once and Debezium resumes from its last committed log position after a crash.

**Gotchas.** Compaction never touches the active segment, so the topic always carries some history. A primary-key update arrives as a delete for the old key and a create for the new one. And with Postgres, unchanged TOASTed columns arrive as a placeholder unless the table uses `REPLICA IDENTITY FULL`.

The reasoning

**acks=0** is fire-and-forget. Latency is lowest and nothing the broker does is reported back, so a lost record is invisible. It suits data where losing some is acceptable and throughput is everything.

**acks=1** waits for the leader to append the record. If the leader dies before a follower has copied it, the record is gone even though the producer was told it succeeded. Consumers never saw it — it was above the high watermark — which is what makes this loss so hard to spot.

**acks=all** (or -1) waits for every replica currently in the ISR. It is the default since Kafka 3.0 and required by the idempotent producer. On its own it can still be one copy when the ISR has shrunk to the leader, which is why it is paired with `min.insync.replicas=2` on a topic with replication factor 3.

The reasoning

The semantics come from two places. On the **producer**, acks=0 is at-most-once, and retries without idempotence can duplicate; the idempotent producer (default since 3.0) removes retry duplicates. On the **consumer**, it is the order of processing and committing: commit first and a crash skips what was not yet processed; process first and a crash re-reads what was already processed.

**Exactly-once** is idempotent producers plus transactions plus `isolation.level=read_committed` on the readers: output records and the input offsets are committed atomically, so a crash replays a whole transaction or none of it. Kafka Streams turns it on with `processing.guarantee=exactly_once_v2`.

Its boundary is Kafka. A consumer that writes to a warehouse, a cache or an API is at-least-once however Kafka is configured, and gets effectively-once only through an idempotent write at the sink — an upsert on a key, or a stored offset in the same transaction as the data.

The reasoning

For each partition a group reads, lag is `log-end offset − committed offset`. Summed over partitions it is the group's backlog. It is approximate on compacted topics and on topics with transaction markers, where offsets are not one per visible record.

Tools: `kafka-consumer-groups.sh --bootstrap-server … --describe --group <g>` for a snapshot per partition; the consumer's own `records-lag-max` metric; and external monitors such as Burrow or a Prometheus exporter that compare committed offsets with log-end offsets over time.

Record counts mislead across topics — a million records is seconds on one and hours on another — so convert lag to **time behind** (the age of the oldest unprocessed record) and compare it with retention: that is how long you can fall behind before data is lost. Alert on lag that keeps growing, not on lag above a threshold, which fires on every harmless spike.

The reasoning

Connect runs as a cluster of workers. You submit a connector configuration through its REST API; the connector splits the work into tasks, the workers run them, and the cluster stores configuration, offsets and status in Kafka topics, so a failed worker's tasks move elsewhere and resume where they left off. Converters handle serialisation (including Schema Registry formats), and single message transforms handle light per-record changes.

It earns its place for plumbing: change data capture, landing topics in object storage, loading warehouses, feeding search. A hand-written consumer for the same job has to reimplement offset tracking, batching, retries, dead-lettering and fault-tolerant scaling — and usually gets one of them wrong.

Write your own when the work is really business logic — enrichment, joins, per-event decisions — which belongs in Kafka Streams, Flink or a service; when no maintained connector exists; or when you need behaviour a connector cannot express. And budget for operating it: a Connect cluster is another distributed system to monitor.

The reasoning

In a **KStream**, every record is a fact: a click, a payment, a status change. Counting a stream of profile updates counts the updates. In a **KTable**, a record is an upsert: a new value for a key replaces the old one, so a table of user profiles holds one current profile per user, and counting it counts users.

A KTable is materialised in a local state store, backed by a compacted changelog topic, and partitioned like its input. A **GlobalKTable** is fully replicated to every instance, which suits small reference data that every task needs to look up.

The distinction drives joins. A **stream-table join** enriches each event with the table's value for its key at the time it is processed, with no window. A **stream-stream join** needs a window, because both sides are unbounded. A **table-table join** produces another table that updates whenever either side changes.