16 · The Producer Client

Source: Apache Kafka 4.4.0-SNAPSHOT (git 04bfe7d, 2026-06-15), KRaft mode. Derived from source code, not copied from official documentation.

The KafkaProducer is an asynchronous, thread-safe client that turns user records into compact, compressed record batches and ships them to partition leaders. A user thread calling send() never touches a socket: it serializes the key/value, picks a partition, and appends the bytes into a per-partition queue inside the RecordAccumulator, returning a Future immediately. A single background Sender I/O thread drains ready batches, groups them by leader broker, builds ProduceRequests, pushes them through the NetworkClient pipeline, and dispatches retries, backoff, and completion callbacks. Idempotence (default-on since 3.0) stamps each batch with a producer id, epoch, and per-partition sequence number so the broker can deduplicate retries while preserving order. This chapter traces the entire path from send() to broker ack, field by field and lock by lock.

Role & responsibilities

The producer client sits at the boundary between application code and the Kafka cluster. Its responsibilities are:

• Serialization & interception. Convert typed keys/values to byte[] via configured Serializers, after optionally rewriting the record through a chain of ProducerInterceptors.
• Partition selection. Honour an explicit partition, else hash the key, else delegate to the adaptive sticky BuiltInPartitioner (or a user Partitioner).
• Batching & buffering. Coalesce records destined for the same (topic, partition) into ProducerBatch objects backed by a bounded BufferPool, trading a little latency (linger.ms) for throughput.
• Transmission. Decide which batches are ready, group them by destination node, encode ProduceRequests, and pipeline them with at most max.in.flight.requests.per.connection outstanding per broker.
• Reliability. Classify responses as success / retriable / fatal; reenqueue retriable batches with exponential backoff; enforce a hard delivery.timeout.ms deadline; and, when idempotence or transactions are on, manage producer ids, epochs, and sequence numbers (see Transactions & EOS).

Where it lives in the code

Core concepts & terminology

RecordAccumulator

The in-memory staging area. A ConcurrentMap<String, TopicInfo> keyed by topic; each TopicInfo holds a ConcurrentMap<Integer, Deque<ProducerBatch>> keyed by partition (RecordAccumulator.java:1301).

ProducerBatch

A single record batch under construction (one MemoryRecordsBuilder) plus a list of thunks (callback + future pairs). Not thread-safe; guarded by the owning deque's monitor (ProducerBatch.java:58).

Thunk

A (Callback, FutureRecordMetadata) pair kept for every appended record so callbacks can fire and so the batch can be split and resent if rejected (ProducerBatch.java:426).

BufferPool

A buffer.memory-bounded allocator. Buffers of exactly batch.size ("poolable size") are recycled on a free list; larger ones are tracked as raw byte counts (BufferPool.java:45).

Sender / I/O thread

The lone background thread named kafka-producer-network-thread | <clientId> that owns all network interaction (KafkaProducer.java:251, KafkaProducer.java:475).

Sticky partition

The current target partition for keyless records, held in an AtomicReference<StickyPartitionInfo> and rotated after roughly batch.size bytes (BuiltInPartitioner.java:48).

Muting

When max.in.flight=1 (so ordering must be exact even without idempotence) the Sender mutes a partition while a batch is in flight to forbid concurrent draining (Sender.java:420).

The send pipeline, end to end

Everything below the public send(record, callback) happens on the calling thread. The interceptor chain runs first and outside the try/catch, so it sees every record even on later failure (KafkaProducer.java:1092); the heavy lifting is in doSend (KafkaProducer.java:1125).

1. Metadata wait

waitOnMetadata returns immediately if the cluster already knows the topic's partition count (and the requested partition is in range); otherwise it adds the topic to the metadata set, calls sender.wakeup() to nudge the I/O thread into issuing a MetadataRequest, and blocks on metadata.awaitUpdate until the topic appears or max.block.ms elapses, at which point it throws TimeoutException (KafkaProducer.java:1244). This is the first of two places send() can block. See Metadata Propagation.

2. Serialization & size validation

Key and value serializers run on the user thread; a ClassCastException is rewrapped as SerializationException (KafkaProducer.java:1150). After serialization, headers are frozen read-only and the upper-bound serialized size is computed via AbstractRecords.estimateSizeInBytesUpperBound. ensureValidRecordSize rejects records larger than max.request.size or buffer.memory with RecordTooLargeException (KafkaProducer.java:1313). The record timestamp defaults to "now" when the user did not supply one (KafkaProducer.java:1176).

3. Partition selection

partition() resolves the target partition in priority order (KafkaProducer.java:1625):

1. If record.partition() is non-null, use it verbatim.
2. Else if a custom partitioner.class is configured, call it (a negative result is an IllegalArgumentException).
3. Else if the serialized key is non-null and partitioner.ignore.keys=false, hash it: Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions (BuiltInPartitioner.java:397). Keyed records are always distributed this way; the adaptive logic never touches them.
4. Else return RecordMetadata.UNKNOWN_PARTITION (the sentinel -1), deferring the choice to the BuiltInPartitioner inside the accumulator.

4. The adaptive sticky partitioner (KIP-794 / KIP-480)

There is one BuiltInPartitioner per topic. It keeps the chosen partition in an AtomicReference<StickyPartitionInfo>; each StickyPartitionInfo carries an immutable partition index and an AtomicInteger producedBytes counter (BuiltInPartitioner.java:381). The append protocol is a careful lock-free peek / lock / re-validate dance (BuiltInPartitioner.java:156):

1. peekCurrentPartitionInfo returns the current sticky info, creating one via nextPartition on first use (CAS-guarded against racing creators).
2. The caller locks the partition's batch deque.
3. isPartitionChanged confirms no other thread switched partitions; if it did, append loops and retries (RecordAccumulator.java:319).
4. Bytes are appended.
5. updatePartitionInfo adds the appended bytes and may rotate the sticky partition.

When to switch. The partition rotates once producedBytes ≥ stickyBatchSize (which is batch.size) and switching is enabled, or unconditionally once producedBytes ≥ 2 × stickyBatchSize (BuiltInPartitioner.java:251). Switching is disabled while the deque has an incomplete (non-full) last batch, so the producer does not strand half-filled 4 KB batches when linger.ms is high, the cap at 2× prevents pathological non-switching (RecordAccumulator.java:327, RecordAccumulator.java:355).

Where to switch. nextPartition picks the next partition. With adaptive partitioning disabled (partitioner.adaptive.partitioning.enable=false) or before stats exist, it chooses uniformly at random among partitions that have a leader (BuiltInPartitioner.java:79). With adaptive partitioning enabled, it draws from a cumulative frequency table weighted inversely by queue depth: each partition's weight is maxQueueSize + 1 − queueSize, folded into a running sum, then a uniform random in [0, total) is mapped via binary search (BuiltInPartitioner.java:270, BuiltInPartitioner.java:111). Partitions with shorter queues, i.e. faster brokers, get proportionally more records.

The stats feeding this table are recomputed by the Sender thread during ready(): partitionReady gathers per-partition deque sizes and calls updatePartitionLoadStats (RecordAccumulator.java:653). If partitioner.availability.timeout.ms > 0, a partition whose leader has not drained data for that long is excluded from the available set entirely, so the partitioner stops sending to a stuck broker (RecordAccumulator.java:729). When partitioner.rack.aware=true, a parallel rack-local CFT is built and preferred when non-empty (BuiltInPartitioner.java:104, BuiltInPartitioner.java:345).

RecordAccumulator: batching & the append algorithm

append is the hot path shared by all user threads (RecordAccumulator.java:280). It increments appendsInProgress (so abortIncompleteBatches on close can't miss in-flight appends) and runs a retry loop:

computeIfAbsent topic → TopicInfo (creates per-topic BuiltInPartitioner)
loop:
  if partition == UNKNOWN: peek sticky partition else use explicit partition
  setPartition(callbacks, effectivePartition)
  dq = topicInfo.batches.computeIfAbsent(partition, ArrayDeque::new)
  synchronized (dq):
     if partitionChanged(...) continue          // racey switch; retry
     r = tryAppend(...) on last batch
     if r != null: updatePartitionInfo; return r // fit into existing batch
  // no room: allocate a fresh buffer (may block on BufferPool up to maxTimeToBlock)
  buffer = free.allocate(max(batch.size, estimatedRecordSize), remainingBlockMs)
  synchronized (dq):
     if partitionChanged(...) continue
     r = appendNewBatch(...)                     // create ProducerBatch, append, addLast
     if r.newBatchCreated: buffer = null          // buffer now owned by the batch
     updatePartitionInfo; return r
finally: free.deallocate(buffer); appendsInProgress--

tryAppend calls ProducerBatch.tryAppend, which checks recordsBuilder.hasRoomFor and returns null when the batch is full; on a full batch it eagerly calls closeForRecordAppends() to release compression buffers (RecordAccumulator.java:430, ProducerBatch.java:147). The buffer is sized as max(batch.size, upperBound(record)), so an over-sized record gets its own larger-than-batch.size buffer (RecordAccumulator.java:334). Note the careful buffer ownership transfer: when a new batch is created the local buffer is nulled so the finally block does not return it to the pool.

What "ready" and "full" mean

A batch counts as full when the deque has more than one batch (so all but the last are sealed) or the last batch's builder reports isFull() (RecordAccumulator.java:719). A node is ready to send if it has at least one non-muted, non-backing-off partition where any of these hold (RecordAccumulator.java:613):

• the batch is full, or
• it has lingered ≥ linger.ms (or, when retrying, ≥ the exponential retry backoff), or
• the BufferPool is exhausted (threads are blocked waiting), or
• the accumulator is closed, or a flush() is in progress, or the transaction is completing.

The RecordAppendResult returned to doSend carries batchIsFull and newBatchCreated; if either is true, doSend wakes the Sender so it does not wait out a full poll timeout (KafkaProducer.java:1192).

BufferPool & back-pressure

The pool is bounded by buffer.memory (default 32 MiB) and guarded by a single ReentrantLock (BufferPool.java:70). Free memory is the sum of nonPooledAvailableMemory plus the recycled free-list count times poolableSize (which equals batch.size). allocate(size, maxTimeToBlockMs):

• returns a recycled buffer instantly if size == poolableSize and the free list is non-empty;
• otherwise, if enough memory is on hand, reserves it (freeing pooled buffers as needed via freeUp);
• otherwise blocks the caller on a per-request Condition appended to a FIFO waiters deque, accumulating memory incrementally as buffers are returned, until it has enough or maxTimeToBlockMs elapses, on timeout it throws BufferExhaustedException (BufferPool.java:159).

The maxTimeToBlockMs passed in is the remaining max.block.ms budget after the metadata wait already consumed part of it (KafkaProducer.java:1146). This is the second place send() can block. The waiter queue is strictly FIFO and memory is handed to the longest-waiting thread first, preventing starvation when a large request must wait for several small buffers to return (BufferPool.java:38). deallocate returns exact-poolableSize buffers to the free list and signals the head waiter (BufferPool.java:260).

ProducerBatch: structure, futures, and the state machine

A ProducerBatch wraps one MemoryRecordsBuilder (the actual on-wire v2 batch, see Record Format & Batches) and a List<Thunk>. Each appended record produces a FutureRecordMetadata chained to the batch's shared ProduceRequestResult; that future is what send() hands back (ProducerBatch.java:147). Key mutable fields: recordCount, maxRecordSize, an AtomicInteger attempts, and an AtomicReference<FinalState> (ProducerBatch.java:64).

The double-completion tolerance matters: an in-flight batch may expire on delivery.timeout.ms (set FAILED) yet still have been written by the broker (a later SUCCEEDED attempt). The CAS-based done() ignores FAILED→FAILED and ABORTED→FAILED, logs the rare FAILED/ABORTED→SUCCEEDED, and throws only on SUCCEEDED→anything (ProducerBatch.java:280). completeFutureAndFireCallbacks sets the future, then invokes every thunk's callback in append order, swallowing exceptions thrown by user callbacks so one bad callback cannot poison the rest (ProducerBatch.java:297).

The Sender thread: drain, encode, transmit, complete

The I/O thread runs Sender.run, which loops runOnce() until running is cleared, then drains residual data and closes the client (Sender.java:240). runOnce first services the TransactionManager (coordinator discovery, InitProducerId, AddPartitions, commit/abort, idempotent epoch bumps), then calls sendProducerData followed by client.poll (Sender.java:310).

sendProducerData

One iteration (Sender.java:380):

1. accumulator.ready(metadataSnapshot, now) → the set of ready leader nodes, the next-check delay, and topics with unknown leaders (which trigger a metadata refresh).
2. For each ready node, gate on client.ready(node, now); nodes whose connection isn't established are dropped this round, and node latency stats are nudged.
3. accumulator.drain(metadataSnapshot, readyNodes, maxRequestSize, now) → Map<nodeId, List<ProducerBatch>>.
4. addToInflightBatches records them for delivery-timeout tracking; if guaranteeMessageOrder (i.e. max.in.flight==1), every drained partition is muted.
5. Expire batches that blew delivery.timeout.ms (both queued and in-flight) via failExpiredBatches.
6. sendProduceRequests encodes and sends one ProduceRequest per node.

drainBatchesForOneNode

For a node, the accumulator walks that node's partitions starting at a per-node round-robin drainIndex (so no single topic-partition starves the others), and for each it pulls the head batch while (RecordAccumulator.java:860): the partition isn't muted; the batch isn't in retry backoff; adding it won't exceed max.request.size (unless the batch is the first and alone); and idempotence ordering permits it (shouldStopDrainBatchesForPartition). When idempotence is on and the drained batch has no sequence yet, it is stamped here: batch.setProducerState(producerIdAndEpoch, sequenceNumber(tp), isTransactional), the per-partition sequence is advanced by recordCount, and the batch is registered as in-flight (RecordAccumulator.java:911). The batch is then close()d (header written, compression finalized) outside the lock because closing is expensive.

Building and sending the ProduceRequest

sendProduceRequest assembles a ProduceRequestData with the configured acks and request.timeout.ms, grouping batches by topic (using topic id where available; modern ProduceRequest v13 drops topic names in favour of ids) and setting each batch's MemoryRecords as the partition payload (Sender.java:896). For transactional producers it stamps the transactionalId and selects the request version. Each batch is marked setInflight(true). The client request is created with expectResponse = acks != 0, for acks=0 there is no response and batches complete as soon as the bytes are written (Sender.java:945, Sender.java:653).

Handling the response

handleProduceResponse demultiplexes per-partition results, resolving the TopicPartition by topic id (v13) or name (older) and dispatching each to completeBatch (Sender.java:580). Disconnects map to NETWORK_EXCEPTION, timeouts to REQUEST_TIMED_OUT, version mismatches to UNSUPPORTED_VERSION. The response may carry updated leader id/epoch which is fed back into the metadata cache for NOT_LEADER_OR_FOLLOWER / FENCED_LEADER_EPOCH errors (Sender.java:638).

completeBatch: the success / split / retry / fail decision

canRetry is the retriability gate (Sender.java:876): the batch must not have reached delivery.timeout.ms, must have attempts() < retries, must not be already done, and the error must be a RetriableException, or, when idempotence is on, must satisfy TransactionManager.canRetry, which additionally retries certain UNKNOWN_PRODUCER_ID and OUT_OF_ORDER_SEQUENCE_NUMBER cases and may request an epoch bump (TransactionManager.java:1043).

Ordering & idempotence

Idempotence is on by default (enable.idempotence=true, ProducerConfig.java:541). When enabled, the TransactionManager is instantiated even with no transactional.id ("an idempotent producer"; KafkaProducer.java:612). On first send the producer fetches a producer id and epoch via InitProducerId (TransactionManager.java:691), then stamps each batch with (producerId, epoch, baseSequence) at drain time. The broker retains the last five batches per producer-partition and rejects duplicates (a retried batch with an already-committed sequence) and gaps (OUT_OF_ORDER_SEQUENCE_NUMBER), giving exactly-once delivery within a producer session.

Reorder-on-retry handling. Once a batch is retried, the accumulator must keep the partition's queue in sequence order. reenqueue adds the batch to the front via insertInSequenceOrder, which linearly scans the in-flight batches and inserts so base sequences stay ascending (RecordAccumulator.java:557). To avoid writing a new-epoch sequence-0 batch ahead of older batches, shouldStopDrainBatchesForPartition stops draining new (sequence-less) batches while the partition has in-flight batches with a stale producer id/epoch, has unresolved sequences, or has an earlier in-flight batch being retried, effectively collapsing to one in-flight request for that partition during recovery (RecordAccumulator.java:823).

Epoch bumps. On a fatal sequence gap an idempotent (non-transactional) producer bumps its epoch locally and restarts sequences at 0 for the affected partition. An OutOfOrderSequenceException from the broker calls requestIdempotentEpochBumpForPartition; on the next runOnce, bumpIdempotentEpochAndResetIdIfNeeded increments the epoch (or, at Short.MAX_VALUE, resets to a brand-new producer id) and rewrites in-flight sequences from the beginning (TransactionManager.java:827, TransactionManager.java:673). An expired in-flight batch also marks the partition's sequence unresolved so no new batch is drained until the gap resolves (Sender.java:373).

acks, timeouts & retries

acks=0

Fire-and-forget. No response is requested; the batch completes when written to the socket and RecordMetadata.offset() is -1. Incompatible with idempotence.

acks=1

Leader-only durability. The leader acks after writing to its log, before followers replicate. Incompatible with idempotence.

acks=all (-1)

The leader acks only after the in-sync replica set has the record (subject to min.insync.replicas). The only value compatible with idempotence; it is the default (ProducerConfig.java:403). See Replication, ISR & High Watermark.

Three timeouts compose (KafkaProducer.java:589): request.timeout.ms bounds one network round trip; retries bounds the number of attempts; and delivery.timeout.ms is the hard wall-clock deadline from when the record enters the accumulator until success or failure is reported, spanning lingering, backoff, and all retries. The constructor enforces delivery.timeout.ms ≥ linger.ms + request.timeout.ms, raising ConfigException if the user set an inconsistent value or silently bumping the default otherwise. Because retries defaults to Integer.MAX_VALUE, in practice delivery.timeout.ms (default 120 s) is what actually terminates retrying, the docs and code both steer users to tune it instead of retries (ProducerConfig.java:402).

Error classification

Compression

Compression is configured once via compression.type (default none; gzip/snappy/lz4/zstd supported, with optional per-codec level) and applied at the batch level by the MemoryRecordsBuilder as records are appended (KafkaProducer.java:562, RecordAccumulator.java:409). The producer maintains a per-(topic, codec) CompressionRatioEstimator so it can size buffers realistically; on close() each batch updates the estimate (ProducerBatch.java:518). Because compression operates on whole batches, larger batches (higher linger.ms/batch.size) compress better. When a compressed batch nonetheless exceeds max.request.size on the broker, the split path resets the estimate conservatively to avoid repeated splits (RecordAccumulator.java:516).

flush() and close()

flush() makes every buffered record immediately sendable and blocks until all in-flight requests complete. It increments flushesInProgress (which makes ready() return everything as sendable), wakes the Sender, then awaits each incomplete batch's ProduceRequestResult, using awaitAllDependents() so split batches are also awaited (KafkaProducer.java:1366, RecordAccumulator.java:1106). Calling flush() from inside a callback throws, because the callback runs on the I/O thread and would deadlock waiting on itself (KafkaProducer.java:1368).

close(timeout) closes the accumulator first (so no new appends are accepted), clears running, and joins the I/O thread, which drains residual batches and completes in-flight requests up to the timeout; if the deadline passes it forceClose()s, failing all incomplete batches via abortIncompleteBatches (KafkaProducer.java:1546, Sender.java:295). Called from a callback, close() degrades to a non-blocking close(0) to avoid a self-join (KafkaProducer.java:1554).

Concurrency & threading model

The design keeps user threads off the network entirely and confines all socket I/O, retry bookkeeping, and the in-flight pipeline to one Sender thread, so the only cross-thread contention is the short critical section around each partition's deque and the BufferPool lock. The hot ready()/partitionReady() loop is deliberately minimal under the deque lock to avoid contending with appends at large partition counts (a documented hot path; RecordAccumulator.java:697).

The NetworkClient pipeline

The Sender sends through a NetworkClient configured with max.in.flight.requests.per.connection. canSendRequest returns true only when the connection is ready, the channel is writable, and InFlightRequests.canSendMore(node) holds; the latter is true when the node has no outstanding requests, or the oldest send has completed and the queue size is below the max (NetworkClient.java:544, InFlightRequests.java:96). This is the mechanism that bounds pipelining per broker. Throttling from quotas is observed as a throttleDelayMs on the connection, which the Sender folds into its poll timeout. The wire encoding of ProduceRequest/ProduceResponse is covered in Wire Protocol & RPC Framework.

Failure modes, edge cases & recovery

• Metadata never arrives. send() blocks up to max.block.ms then throws TimeoutException; if a partial wait already elapsed, the BufferPool gets the remainder (KafkaProducer.java:1146).
• Buffer exhausted. allocate blocks up to the remaining max.block.ms, then throws BufferExhaustedException (a subclass of TimeoutException), incrementing buffer-exhausted-records (BufferPool.java:159).
• Delivery timeout. A batch that exceeds delivery.timeout.ms in queue or in flight is failed with TimeoutException regardless of remaining retries (Sender.java:362). With idempotence this triggers a sequence-unresolved marker and eventual epoch bump.
• Leader change. NOT_LEADER_OR_FOLLOWER/FENCED_LEADER_EPOCH are retriable; the response's new leader info updates metadata, and the batch tracks the new leader epoch so backoff is skipped when retrying to a fresh leader (ProducerBatch.java:116, RecordAccumulator.java:804).
• Forced close. abortIncompleteBatches loops until appendsInProgress == 0 so no batch from a racing appender is missed, then aborts all incomplete batches (RecordAccumulator.java:1135).
• Batch still in network use (KAFKA-19012). A batch sent to the NetworkClient may have its pooled buffer in use; deallocation is deferred (maybeRemoveAndDeallocateBatchLater) until the request truly completes, to prevent reusing a buffer that the selector is still reading (Sender.java:854, RecordAccumulator.java:1168).
• Authorization mid-transaction. TransactionalIdAuthorizationException/ClusterAuthorizationException fail pending requests and transition the manager to UNINITIALIZED so the user need not reconstruct the producer (Sender.java:351).

Invariants & guarantees

Configuration reference

Interactions with other subsystems

• Record Format & Batches, the v2 batch the MemoryRecordsBuilder emits, including producer id/epoch/baseSequence header fields.
• Wire Protocol, ProduceRequest/ProduceResponse schemas and versioning (topic ids in v13).
• Replication, ISR & HWM, what acks=all and min.insync.replicas actually require on the broker.
• Transactions & EOS, the TransactionManager state machine, AddPartitions, and the consume-transform-produce pattern via sendOffsetsToTransaction.
• Metadata Propagation, how the producer learns partition leaders and reacts to leader changes.
• Quotas & Throttling, produce throttling surfaced as connection poll delays.
• Network Layer and Request Processing, the broker side that the Sender talks to.

Design rationale & evolution

Gotchas & operational notes