Producers: Advanced

A trip ends and the fare service sends 2 records: the fare to fares and the driver's payout to driver-payouts. Those are 2 topics, so 2 partitions with 2 leaders, usually on 2 brokers. An idempotent producer promises that each record lands once in its own partition and nothing more: a crash after the first ack and before the second send leaves a fare with no payout. A transaction ties the 2 sends together. Every record sent between begin_transaction() and commit_transaction() becomes committed together, or none of them does.

Setting transactional.id is the switch. The client refuses to start unless idempotence is on, acks=all is set and at most 5 requests are in flight, because a transaction is built on the idempotent producer's id and sequence numbers. The id names a role and stays the same across restarts.

init_transactions() runs once per producer. It asks any broker where the transaction coordinator for fare-writer-7 is: the coordinator is the leader of the partition of the internal topic __transaction_state that the id hashes to, a normal broker with a normal replicated log. The coordinator assigns producer id 4001 at epoch 0, writes that assignment to its log, and aborts anything an earlier run under the same id left open. begin_transaction() sends nothing: it flips local state. The first produce() to a partition appends the record there and, on a 4.x cluster, that partition's leader tells the coordinator it now belongs to the open transaction. By the time commit_transaction() runs, the coordinator knows the transaction touched fares-2 and driver-payouts-5.

commit_transaction() is a 2-phase write. Phase 1: the coordinator appends a PrepareCommit record for producer 4001 to __transaction_state, a topic created with a minimum in-sync count of 2, so the decision is on 2 disks before anything else happens, and the producer's call returns at this point. Phase 2: the coordinator sends a COMMIT marker to the leader of every partition in the transaction. Each leader appends the marker as a control record, a record with no key and no value that takes 1 offset of its own. When every marker is acknowledged, the coordinator appends CompleteCommit and the transaction is over.

abort_transaction(), or a crash before commit, ends in an ABORT marker written the same 2-phase way: PrepareAbort in the coordinator's log, then a marker in each partition. The fare at offset 40 stays where it is. The log is append-only, so an abort cannot pull a record back; it can only add a marker saying that 40 does not count. Retention deletes the aborted record on the usual schedule, and compaction discards it when the cleaner reaches its segment.

One transaction can hold any number of records on any number of partitions and topics, at the cost of 1 marker per partition it touched. A transaction per record on a hot topic doubles the writes to that partition; a transaction per batch of 500 records adds 1 marker. transaction.timeout.ms, 60 seconds by default, bounds how long a transaction may stay open: the coordinator aborts it on its own once the clock runs out, so a producer that waits on a slow external call mid-transaction finds its commit rejected.

The fare writer runs under an orchestrator that restarts anything that stops answering health checks. Process A owns transactional.id=fare-writer-7, holds producer id 4001 at epoch 0, and has a transaction open with the fare for trip-9001 already at offset 40 of fares-2. Then A stalls for 40 seconds: a garbage collection pause, or a network link that drops. The orchestrator stops hearing from A and starts process B from the same config. Now 2 processes each believe they are fare-writer-7, and A will wake up with sends still queued.

B calls init_transactions(). The coordinator looks up fare-writer-7, finds producer id 4001 at epoch 0 with a transaction ongoing, bumps the epoch to 1, hands B the pair (4001, 1), and aborts A's open transaction: an ABORT marker goes to every partition A had registered. The coordinator's log now says epoch 1 is current, and each partition leader learns the new epoch from the marker it appended.

Every produce request carries the producer id and the epoch. The leader of fares-2 already keeps per-producer state for idempotence, the last epoch and sequence number it accepted for 4001, and the marker at offset 41 moved that epoch to 1. A's request at 41 seconds arrives with epoch 0. The leader rejects it as stale, and the client raises ProducerFencedException, which is fatal: the producer can no longer retry, open a new transaction or commit. Trip-9002 from A never enters the log, and a reader of committed records sees trip-9001 exactly once, at offset 42, because offset 40 sits under an ABORT marker.

Fencing depends on 1 rule: at any moment, 1 live producer per transactional.id. The rule holds only if the id describes something there is 1 of. A service that runs 8 replicas and gives every replica transactional.id=fare-writer breaks it: each replica's init_transactions() bumps the epoch past the other 7, every replica is fenced on its next send, and the service writes nothing. The scheme that works gives each unit of ownership its own id. With 12 shards of trips, the ids are fare-writer-0 through fare-writer-11, and whichever process owns shard 7 opens its producer with fare-writer-7. On a restart the new owner takes the id, the coordinator bumps the epoch, and the old owner, if it is alive anywhere, is fenced on its next request.

An idempotent producer without a transactional.id gets none of this. It asks for a producer id with no name attached, receives a fresh one at epoch 0, and the broker deduplicates its retries against that id alone. 2 instances of it write freely into the same partition. Fencing is a property of the transactional id, because the id is the only thing 2 processes can be made to share.

The rule from the beginner lesson was: a keyed record goes to hash(key) mod partitions. That rule does not say which hash. The Java client's built-in partitioner runs murmur2 over the key's serialized bytes, clears the sign bit with & 0x7fffffff, and takes the remainder by the partition count. librdkafka, the C library underneath the Python, Go, .NET and Node clients, ships with partitioner=consistent_random, which runs CRC32 over the same bytes and takes the remainder. Both functions are deterministic and spread keys evenly, and they are different functions, so a topic written by a Java service and a Python service with the same keys ends up with each key in 2 partitions, 1 per library.

Take user-1. The Java client computes murmur2 of the 6 bytes of user-1 and gets -1118138342; clearing the sign bit gives 1029345306, which divides by 6 exactly, so partition 0. librdkafka computes CRC32 of the same 6 bytes and gets 1393466234, remainder 2, so partition 2. The 2 results agree for a key with probability 1 in 6, and none of the 6 keys above agrees. Each library is internally consistent: every user-1 record from the Java service lands on 0, every one from the Python service on 2, and a reader of partition 0 sees half of user-1's history in order with no sign that the other half exists.

The bytes matter as much as the function. Both libraries hash the serialized key, so 2 producers using the same library still disagree if 1 serializes the user id as the string 42 (2 bytes) and the other as a 4-byte integer. A topic's partitioning contract is 3 things together: the key's bytes, the hash function and the partition count, and any producer that changes 1 of the 3 is writing to a different place.

librdkafka's murmur2_random reproduces the Java client's function bit for bit, including the sign-bit mask and the random spread for records with no key, so a Python producer with that 1 line puts user-1 on partition 0 next to the Java service. The reverse is possible too: partitioner.class on the Java side accepts a class that computes CRC32, which someone has to write. The decision is per producer, so a mixed fleet needs every producer configured the same way, settled before the first record, because every record already written stays where it landed.

The _random suffix names what the partitioner does with a record whose key is null: consistent_random and murmur2_random pick a random partition for keyless records, while plain consistent and murmur2 hash the empty key and send every keyless record to 1 partition. The Java client sticks keyless records to 1 partition until a batch fills or lingers out, then moves on. None of this touches keyed records, so a mismatch between libraries shows up only where order matters.

A topic with 3 partitions is carrying more than it should, so the operator runs 1 command: kafka-topics.sh --bootstrap-server broker1:9092 --alter --topic events --partitions 6. The command returns in under a second and nothing moves: the 3 old partitions keep every record they hold, and 3 empty partitions appear. What changes is the arithmetic for every future send. hash(key) mod 3 becomes hash(key) mod 6, and for about half of all keys that is a different number.

Follow user-3. Before the change its 4 records sit on partition 1 at offsets 200 to 230. After the change, its next record lands on partition 4 at offset 0. Every record user-3 ever had is still in the topic, but its history is now split across 2 partitions, and the per-key order the beginner lesson promised, 1 key on 1 partition read in offset order, holds within each half and not across them. A reader that is behind on partition 1 and current on partition 4 sees the newest record first. A reader that never looks at partition 4 sees nothing new for user-3 at all.

Compacted topics make the split permanent. A compacted topic of account balances keeps the latest record per key per partition, because the cleaner works 1 partition at a time and never looks across. After user-3 moves, partition 1 keeps balance=120 as its latest and partition 4 keeps balance=95 as its latest, and both survive every compaction pass forever. A reader rebuilding a table from the topic sees 2 latest values for 1 key and keeps whichever it reads last. The clean path for a compacted topic is a new topic with the new count, refilled from the old one.

A producer learns the partition count from its metadata and refreshes it every metadata.max.age.ms, 5 minutes by default, or sooner when a send fails. 2 producers that write the same keys refresh on their own clocks. For up to 5 minutes after the command, producer P1 may send user-3 to partition 4 while producer P2 still sends it to partition 1, so the split interleaves both ways during the window. A transaction does not protect against this: it makes records commit together on whichever partitions the producers computed.

send() returns at once because it never talks to a broker. It hands the record to an in-memory accumulator, sized by buffer.memory, 32 MiB by default, and the producer's single I/O thread drains batches from there to the leaders. The application fills, the brokers drain, and when the fill outruns the drain the accumulator grows until it reaches buffer.memory. Then send() stops returning at once.

The accumulator is full 8 seconds after the loop starts. The next send() cannot allocate a batch, so it blocks the calling thread until a batch is acknowledged and its memory comes back. From that point the application runs at 6,000 records a second, because the brokers set the pace. That blocking is the producer's backpressure, and it works as long as the brokers keep draining. The wait has a limit: max.block.ms, 60 seconds by default. A send() that waits that long throws TimeoutException, whose message says it failed to allocate memory within the configured max blocking time, and the record it was handed never entered the buffer.

The buffer is 1 pool shared by every partition. A batch for a partition whose leader is down stays in the accumulator while the producer retries it, up to delivery.timeout.ms, 120 seconds by default. With 12 partitions and 1 dead leader, the dead partition's batches stop being freed while the other 11 keep draining, so its share of the pool grows until its batches expire and the memory returns. Whether the pool fills first depends on the rate. With keys evenly spread over the 6,000 records a second the brokers were draining, at 1,024 bytes each, the dead partition's twelfth is 500 records a second. Holding 120 seconds of them would take 60,000 records, about 59 MiB, nearly twice the 32 MiB pool, so the pool fills after about 65 seconds, long before the first batch expires. From then on a send() to a healthy partition blocks as well, because memory has no partition, and 1 slow partition has stalled the whole producer.

Raising buffer.memory is the first instinct and the weakest fix. Doubling it to 64 MiB delays the block from 8.2 seconds to 16.4 seconds, after which the application runs at the drain rate anyway: the buffer sizes a burst, never a rate. Rate fixes live on the drain side: compression shrinks what crosses the wire, a longer linger.ms fills larger batches, and more partitions on more brokers add leaders to write in parallel; each raises the 6,000. On the fill side, the application can read the block as a signal: watch the producer's bufferpool-wait-ratio metric and slow the loop that calls send() when it climbs. A max.block.ms of 2,000 turns a silent minute into an exception the application can act on by shedding the record, spilling it locally or pausing its intake.

The 2 libraries hand the same problem to the application in 2 ways. The Java client absorbs it as time: the thread stops. librdkafka returns it as an error: the call fails and the record is still the application's. A working librdkafka loop polls for delivery reports to free queue slots, then calls produce() again; a working Java loop catches TimeoutException and does the same. Either way the refused record is lost unless the application keeps it, and delivery.timeout.ms never starts for it, because that clock begins when a record enters the buffer.