Two waits happen inside send(), and both are bounded by the same setting, max.block.ms (default 60000, verified in the Kafka 4.0 producer configuration reference). The first is metadata: the very first send to a topic must learn its partitions, and a typo in the topic name or a broker that is unreachable shows up as TimeoutException: Topic orders not present in metadata after 60000 ms. The second is memory: if the Sender cannot drain batches as fast as the application fills them, the pool of buffer.memory (default 32 MiB) runs dry and send() waits for space.

• A ProducerRecord has a topic, an optional partition, an optional key, a value, optional headers, and an optional timestamp (default: the time of send(), CreateTime).
• The key decides the partition (page 02, chapter 07) and therefore the ordering group. The value may be null: on a compacted topic that is a tombstone (delete marker).
• Headers are key and byte-value pairs that travel with the record and are never used for partitioning. Use them for trace ids, event ids for deduplication, content type, and schema hints.
• KafkaProducer is thread-safe. Create one instance per application (per set of settings) and share it. One producer per request or per thread wastes connections and defeats batching.

The partitioner runs in the caller thread, before the record reaches the accumulator. In Kafka 4.0 the producer has a built-in default and partitioner.class is null (verified); you only set it to plug in your own logic.

gzip ships with the JDK, but snappy, lz4 and zstd are implemented by separate libraries that kafka-clients declares as optional. If you set compression.type=lz4 without org.lz4:lz4-java on the classpath, the producer fails when it first compresses a batch. This is a real failure we hit while preparing this page: NoClassDefFoundError: net/jpountz/lz4/LZ4Factory. Spring Boot starters and the Confluent bundles usually pull these in; a plain kafka-clients project does not.

The last row is the one to remember. A timeout is not "the message was not written". It means "no acknowledgment arrived in time". Without idempotence, resending after such a timeout in your own code can create a duplicate; this is exactly the ambiguity the next chapter resolves.

Records of one partition are written in the order batches reach the broker. With several batches in flight on one connection, a failure of an early batch followed by success of a later one lets the retry land after its successor (lower panel of the diagram). That is why older guides say "set max.in.flight.requests.per.connection=1 for strict order", at a large throughput cost.

With idempotence enabled (the default since Kafka 3.0) that advice no longer applies. The broker tracks a sequence number per producer and partition and rejects a batch that arrives ahead of its predecessor, so ordering is preserved for any in-flight value up to 5. The producer configuration reference states this directly: with idempotence on, ordering is preserved, and the value must be 5 or less because the broker retains metadata for at most five batches per producer.

Kafka 4.0 also removed a convenience: the upgrade notes say enable.idempotence no longer silently falls back when max.in.flight.requests.per.connection exceeds 5. If you set idempotence explicitly with conflicting values (an in-flight above 5, acks other than all, retries=0) the producer throws a ConfigException.

Every batch stored by the broker carries the producer id, epoch and sequence. kafka-dump-log.sh prints them. Below are three batches from partition 0 of the lab topic in chapter 10, abridged (extra columns removed from lines that were produced by a real Kafka 4.0.0 broker). The first two came from the Java producer of chapter 10 (PID 2, sequences 0 and 1), the third from kafka-console-producer.sh (a different PID, so its sequence starts again at 0).

Idempotence therefore protects one producer session against its own retries. Surviving a restart needs a stable identity, which is what transactional.id provides (next chapter), or an application-level event id that the consumer deduplicates on (page 01, idempotent consumers).

Data is appended to the log as soon as it is sent, before the transaction commits. The broker therefore tracks the last stable offset (LSO): the offset below the first record of any still-open transaction. A read_committed consumer can read only up to the LSO and the broker sends it the list of aborted ranges so it can skip them. Consequences:

• Latency of downstream consumers includes the transaction duration: nothing after an open transaction is visible until it completes, even records of other producers (the consumer reads in offset order).
• A hung transaction stalls every read_committed consumer of that partition until transaction.timeout.ms expires. The consumer docs say so directly: in read_committed mode poll() returns only up to the LSO, so consumers cannot reach the high watermark while a transaction is in flight.
• Commit and abort markers are control records. They occupy an offset but are never returned to the application.

You can see the markers in a real run. The lab in chapter 10 reads 35 records from orders and writes 35 transformed records to orders-enriched, using TransformRelay from chapter 10. Yet the log end offsets of the output topic add up to 38, three more than the input:

The extra three offsets are the commit markers, one per partition written. kafka-dump-log.sh shows the batch (transactional, with PID 4) followed by the control batch (abridged below). Note the epoch moves from 1 to 2 between the data and the marker: with Kafka 4.0 clients and brokers the producer epoch is bumped as part of every transaction (KIP-890, "server side defense", described in the 4.0 upgrade notes), so a delayed message from an earlier transaction can never be mistaken for the current one.

"Exactly-once" in Kafka means: for a read-process-write cycle whose inputs and outputs are all Kafka topics, each input record's effect appears exactly once in the outputs. That is a precise and useful claim, and a narrow one.

Everything a consumer does is driven by one call, poll(Duration), in a loop run by one thread. A single poll() can fetch records, send heartbeats and join or leave the group (on the classic protocol a background thread also sends heartbeats), run rebalance callbacks, and, if enabled, auto-commit offsets. If you stop calling it, the consumer stops being a group member.

A consumer group is a set of consumers sharing one group.id. Kafka assigns each partition of the subscribed topics to exactly one member, so the members split the work and never process the same partition at the same time. If a member dies its partitions are given to the survivors. That is how a consumer application scales out and fails over.

• Each group has a group coordinator: the broker that leads the __consumer_offsets partition the group id hashes to (50 partitions by default, offsets.topic.num.partitions). It tracks membership, heartbeats and commits.
• On the classic protocol the group's members send their subscriptions to the coordinator, one member (the group leader) computes the assignment on the client with an assignor, and the coordinator distributes it. On the new protocol (KIP-848, chapter 08) the broker computes it.
• subscribe(topics) uses group management and rebalancing. assign(partitions) skips it: you name the partitions yourself, there is no group membership and no rebalance (used by replay tools, chapter 07).

The Kafka 4.0 default is partition.assignment.strategy = [RangeAssignor, CooperativeStickyAssignor] (verified). A group uses the first assignor that all members support, so by default it runs Range, which is eager. The list exists to let you move to cooperative with one rolling restart that drops RangeAssignor. We saw this in the lab: kafka-consumer-groups.sh --describe --state printed ASSIGNMENT-STRATEGY range for a default classic group.

• Commits are written to an internal topic, __consumer_offsets (50 partitions, replication factor 3 by default). A commit is a normal record: key = (group, topic, partition), value = offset plus metadata and leader epoch.
• The topic is compacted: only the latest commit per key is kept. The group's coordinator is the leader of the partition its id hashes to.
• Committed offsets of a group with no members are retained for offsets.retention.minutes (default 10080, 7 days) and then deleted (Kafka 4.0 broker configuration reference).
• Consumers do not have to use Kafka for offsets at all (see the database row in the table below), but the group tooling and lag monitoring read from here.

This setting is not "where to start every time". Once a group has a committed offset, it is ignored. To start somewhere else you must reset the group's offsets or seek().

• seek(partition, offset), seekToBeginning(...), seekToEnd(...) set the position. The next poll() fetches from there.
• offsetsForTimes(Map<TopicPartition, Long>) returns, for each partition, the earliest offset whose timestamp is at or after the given time (or null if no such record). Combine it with seek() to "replay from yesterday 09:00". It uses the .timeindex files of page 02.
• beginningOffsets() and endOffsets() give the first retained and the next-to-be-written offsets.
• From outside the application: kafka-consumer-groups.sh --reset-offsets with --to-earliest, --to-latest, --to-offset, --to-datetime, --by-duration, --shift-by, or --from-file. The group must be inactive (chapter 10 shows the exact error).

Replay is the log's superpower, and the reason consumers must be idempotent (below): a replay deliberately delivers old records again.

Lag is the number of records between the log end offset and the group's committed offset, per partition. It is the one consumer-health metric to alert on: a steadily growing lag means the consumers are slower than the producers.

• Lag in records is not lag in time. Ten thousand records may be seconds or hours behind, depending on the rate; Kafka does not report time lag. Estimate it from rate, or compare record timestamps.
• A lag that grows only on one partition points at a hot key or a poison record, not at capacity.
• A lag of zero with a stalled consumer is possible if nothing is produced: pair lag alerts with a "consumer is polling" check.

Everything on this page produces at-least-once delivery by design: a crash, a rebalance, or a replay re-delivers records that may already have had effects. Page 01 introduced the cure, the idempotent consumer: make processing a record twice equal to processing it once. The robust form puts the dedup record and the business write in one database transaction; chapter 10 has the Java.

• Dedup key: prefer a business event id that the producer put in the payload or a header. topic-partition-offset also works for plain replays, but it changes if the producer re-sends the same event (a new offset) and is meaningless after copying to another cluster.
• Order matters: process and commit the DB transaction first, then commit the Kafka offset. A crash in between replays the record, and the dedup row turns the replay into a no-op.
• Retention of dedup rows: keep them at least as long as a replay can happen (the topic retention) plus the maximum consumer downtime.

The two timeouts exist because they detect different failures: heartbeats prove the process and network are alive; the poll interval proves the application loop is making progress. A consumer stuck in a slow database call keeps heartbeating but stops polling, and the poll interval is what evicts it.

In an eager rebalance every member revokes all its partitions, rejoins, and receives a new assignment, even if most partitions end up with the same owner. During that window the whole group is stopped. With a big group, state to rebuild, or a rolling deploy that triggers one rebalance per pod, the pauses add up.

Cooperative rebalancing (CooperativeStickyAssignor) is incremental: in a first round only the partitions that must move are revoked; in a second round they are assigned to their new owners. Everything else keeps being consumed. To adopt it set partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor (or, on the default list, drop RangeAssignor after all members have been upgraded to a version that lists both).

Normally a restarted consumer gets a new member id, so a restart means two rebalances (leave, join). Setting a unique, stable group.instance.id per instance (default null) makes the member static: if it comes back within session.timeout.ms, the coordinator hands it the same partitions and no rebalance happens. The price: when a static member truly dies, its partitions sit idle until the session timeout, so pair it with a session timeout that matches your restart time (for example 1-2 minutes in Kubernetes with a StatefulSet name as the instance id).

Kafka 4.0 shipped a redesigned group protocol. What the 4.0 documentation says, and what we saw in the lab:

Migration is online: roll your consumers out with group.protocol=consumer. When the first such member joins, the group converts from classic to consumer and interoperates; this requires that the classic group uses an assignor that does not embed custom metadata. A group is converted back when the last new-protocol member leaves.

We confirmed the behaviour on a Kafka 4.0.0 broker with the 4.0.2 Java client. Two consumers using group.protocol=consumer produced ASSIGNMENT-STRATEGY uniform in --describe --state, and in the verbose member listing the CURRENT-EPOCH and TARGET-ASSIGNMENT columns are populated (they are empty - for a classic group). When the second consumer joined, the first one received only a revoke of the partitions that moved (revoked [orders-1, orders-2]) and kept reading orders-0: an incremental hand-over, with no stop-the-world step.

A custom format is a class implementing org.apache.kafka.common.serialization.Serializer<T> (serialize(topic, data)) and Deserializer<T> (deserialize(topic, data)), configured with key.serializer / value.serializer on the producer and the deserializer pair on the consumer. Kafka ships String, Integer, Long, Double, UUID, ByteArray and a few more.

Without one, a producer team renames total to amount and every consumer breaks at runtime. A schema registry is a small service (the best-known one is the Confluent Schema Registry; Apicurio and AWS Glue Schema Registry are alternatives; it is not part of Apache Kafka) that stores schemas by subject, assigns each version a global id, and enforces compatibility rules before accepting a new version. The record carries only the id.

With the default subject naming (TopicNameStrategy) the value schema of topic orders is registered under the subject orders-value.

The mode answers "who may be on the old schema while the other side moves to the new one?". Names and rules below follow the Confluent Schema Registry documentation (checked 2026-10-31); the default mode is BACKWARD.

Why BACKWARD is the default: it lets you rewind a consumer to the beginning of a topic and still read everything with the newest schema. That is the replay scenario of chapter 07. (The same documentation notes that for Protobuf the recommended mode is BACKWARD_TRANSITIVE, because adding message types is not forward compatible, and that Kafka Streams supports only BACKWARD.)

A consumer already upgraded to version 2 reading a version 1 record fills currency with the default EUR. Adding the field without a default would be rejected by the registry in BACKWARD mode, because the new reader would have no value for old data.

To compile without Maven, download the three jars and compile all samples of this chapter into one directory. The lab commands below assume $CP is set as shown.

OrderProducer states the durable settings explicitly (even though they are the defaults), sends five keyed records with a callback, and closes cleanly. Note what is missing: no flush() per record, no future.get(), no new producer per message.

• try-with-resources calls close(), which waits for in-flight batches, so records are not lost at exit. A bare System.exit() after send() does lose buffered records.
• The callback runs on the Sender thread. Do not block there; hand heavy work to another executor.
• Compression is commented out on purpose: lz4, snappy and zstd need their own jar (chapter 02).

OrderConsumer is the loop most applications should start from: auto commit off, process the batch, then commitSync(); wakeup() from a shutdown hook to exit poll(); a rebalance listener that commits before partitions are taken away.

• commitSync() with no arguments commits the position of every assigned partition, that is, the offset after the last record poll() returned. If process() throws midway, the loop dies before the commit, and the batch is replayed: at-least-once.
• onPartitionsRevoked commits the work done for partitions about to leave. Without it, a rebalance replays up to one batch for those partitions.
• wakeup() is the only KafkaConsumer method that may be called from another thread. It makes the blocked poll() throw WakeupException, which we treat as a normal shutdown.
• auto.offset.reset=earliest applies only to a group with no committed offsets.

TransformRelay is the loop of chapter 05 in code. The consumer has auto commit off and reads read_committed; offsets travel with the transaction through sendOffsetsToTransaction(offsets, consumer.groupMetadata()); the committed offset for each partition is the next offset to read, record.offset() + 1.

assign() instead of subscribe() means this tool is not a group member: no rebalances, and it will not disturb the real group. If offsetsForTimes returns null for a partition (no record at or after the time), the example seeks to the end.

The handler inserts the dedup row and applies the change in one database transaction. It uses only java.sql, so it compiles with no driver; run it against any database that supports ON CONFLICT DO NOTHING (PostgreSQL) or adapt the insert (INSERT IGNORE on MySQL).

Call handle(record) in the loop of OrderConsumer, and commitSync() only after it returned. A crash between the database commit and the Kafka commit replays the record, and the dedup row makes the replay harmless.

Everything below was run on a single-node KRaft broker from the official image (the image's defaults advertise localhost:9092, so clients on your machine connect directly). Output blocks are real but abridged: the HOST and CLIENT-ID columns and some whitespace were trimmed, and long consumer ids shortened. Run docker rm -f kafka first if the container from page 02 is still there.

Keys customer-0 and customer-1 hash to different partitions, so offsets are per partition, exactly as in page 02. Now look at what Kafka stored for the group:

CURRENT-OFFSET is the committed offset (the next record to read); it equals LOG-END-OFFSET, so lag is zero. The group has no members, yet its offsets are kept (up to offsets.retention.minutes). Next, produce 30 more records while no consumer is running, to create lag:

Lag per partition is LOG-END-OFFSET - CURRENT-OFFSET: 9, 12 and 9 records waiting. Start two copies of OrderConsumer (two terminals) and inspect the live group. Then try to reset offsets while they run:

Three things to notice. The group runs the range assignor (the 4.0 default list starts with RangeAssignor). One member got two partitions and the other one, as 3 partitions cannot split evenly. And Kafka refuses to reset offsets for a group with live members. Stop the consumers, then preview with --dry-run (nothing changes), apply with --execute, and verify:

Relative resets are clamped to the valid range, which is what the broker told us when we tried --shift-by -5 on a group positioned near the start of the log:

Other reset options follow the same pattern: --to-latest, --to-datetime 2026-10-31T09:00:00.000, --by-duration PT1H, --shift-by N, --from-file plan.csv. Always run --dry-run first.

To see the new rebalance protocol, run the same consumer with one extra line, props.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, "consumer"), and a different group.id (here billing848), twice:

The strategy column reads uniform and the epoch columns are filled in. Start the first consumer alone, then the second: the first logs a revoke of only the two partitions that moved and keeps reading the third. A classic group shows empty (-) epochs and, with Range, would have revoked all three.

Finally, the transactional relay of chapter 05. It reads orders (35 records) and writes the transformed values to orders-enriched:

The first start can take up to half a minute while the broker creates its internal __transaction_state topic. Restart the relay and nothing is processed twice: the committed input offsets moved inside the transaction.

Committing first is attractive because it avoids replays, but it trades them for loss, which is almost always worse. If you truly need at-most-once (a metrics sampler, say), do it deliberately and document it.

Fixes, from cheapest: lower **max.poll.records so a batch fits the interval comfortably; raise max.poll.interval.ms only to cover legitimate variation; speed up or time-limit the downstream call; use cooperative or the KIP-848 protocol to make each rebalance cheap; add static membership so rolling deploys do not churn; and, if the work is genuinely long, decouple** it: poll in one thread, hand records to a worker pool, call consumer.pause(partitions) while the pool is busy (a paused consumer keeps polling and heartbeating but returns no records) and commit only offsets that the workers have finished, in order.

KafkaConsumer is deliberately not thread-safe. When two threads enter it at the same time, the second gets this exception (reproduced in the lab with two threads polling one consumer):

The rule is one consumer per thread, all with the same group.id; Kafka splits the partitions among them. wakeup() is the single exception. ConsumerPerThread shows the pattern, and the thread count should not exceed the partition count.

The producer is the opposite: KafkaProducer is thread-safe and should be shared.

• **A new KafkaProducer per message or request.** Each instance opens connections, fetches metadata and has its own tiny batches. Create one and reuse it (Spring and your DI container make this a singleton bean).
• **send(...).get() for every record.** It turns the pipeline into one request at a time. Use the callback, and get() only where you truly need synchronous confirmation.
• **flush() after every send().** Same effect: no batching.
• Ignoring the callback. A silent TimeoutException after 120 s is a lost record nobody noticed.
• Copying an old configuration. max.in.flight.requests.per.connection=1, retries=3, acks=1 appear in years-old tutorials. Today's defaults (acks=all, idempotence on, 5 in flight, retries bounded by delivery.timeout.ms) are better and safer.
• No key where order matters, or a key with very few distinct values (hot partition). See page 02, chapter 07.
• **Exiting without close().** Buffered records never leave the process.

• Auto commit with asynchronous processing. The next poll() commits offsets of records another thread has not finished. A crash loses them. Disable auto commit as soon as processing leaves the polling thread.
• Retrying a poison record in place with an infinite loop. The partition stops. Bound the retries, then send the record to a dead-letter topic and move on (page 07 covers retry and DLQ designs).
• **A heavy onPartitionsAssigned** (loading large state) that blocks the poll thread past the poll interval.
• Committing offsets for partitions you no longer own after a rebalance; the commit fails, and the work is duplicated by the new owner. Commit in onPartitionsRevoked.
• Treating a group id as a name for "the topic" and reusing it across unrelated applications; they will split the partitions between them and each sees only part of the data.