Digging into Kafka's Source 5: Inside the Producer — From send() to Batches and the Sender Thread

What this post covers

This is the fifth post in the "Digging into Kafka's Source" series. In part 3 we looked at the byte shape of requests and responses, and in part 4 we followed those bytes from their arrival at a broker to KafkaApis.handle(). Now we cross over to the sending side: from the moment an application calls producer.send(record), to that record leaving through a socket inside a Produce request, to the response coming back and the callback being called.

A close-up of the producer inside area A (clients) of the part 2 map. Blue boxes are threads and processing steps, yellow boxes are memory, and grey boxes belong to the posts before and after this one.

There are four questions to answer.

• What happens on the thread that calls send(), and what does it get back — and when does send() block?
• Which partition does a record without a key go to?
• In which if are linger.ms and batch.size used, and how do batches really change when you change them?
• Which thread actually sends the records, which thread runs the callback, and how many times — and until when — is a failed batch retried?

All line numbers in this post refer to this commit. Part 1 shows how to check out the same commit and build the Docker image used for the examples. The producer code we read follows the same rules as 4.4.0, the latest stable release most people run. BufferPool.java, ProducerConfig.java and InFlightRequests.java are identical to the 4.4.0 tag to the character, and RecordAccumulator.java only has a clean-up (method arguments changed to TopicPartition, a few helper methods added); the conditions this post talks about (ready(), drain(), append()) are the same. The differences in KafkaProducer.java are mostly on the side of two-phase commit for transactions.

The producer is large even by the standards of the clients module: KafkaProducer.java has 1,837 lines, RecordAccumulator.java 1,500 and Sender.java 1,144, and transaction and idempotence handling live in the same files. This post picks only the main path that a first-time reader can follow; what was left out is listed by name and location in "What this post leaves out" at the end.

Terms to know first

Record batch: a bundle of several records going to the same partition. One header (base offset, record count, producer id, CRC …) is followed by the records. The producer does not send records one by one but in these batches, and the broker writes them to disk in the same units. That is why opening a segment file (one of the disk files a broker splits a partition's log into, part 7) with kafka-dump-log.sh shows how many batches there are and how many records each holds (the examples in this post count them that way). The byte layout of a batch is the subject of part 6.

RecordAccumulator: the producer's "waiting room for records to send". It keeps a queue of batches (Deque<ProducerBatch>) per partition and appends each record you send() to the last batch in that partition's queue. To accumulate means to pile up.

Sender thread: a background thread created when you construct the producer, named kafka-producer-network-thread | <client.id>. It takes the batches worth sending out of the waiting room, groups them into one Produce request per broker, sends them, and when responses arrive completes each record's Future and calls the callbacks. Application threads never touch the socket.

linger.ms and batch.size: the two settings that decide when a batch is sent. batch.size is the target size of one batch (bytes, default 16384); linger.ms is how long a batch may wait before it is sent even if it is not full (default 5 milliseconds). Both appear verbatim in the if of section 4.

Metadata: for each topic, how many partitions it has and which broker leads each partition. A producer can only write to the leader, so it needs this first to decide which broker a batch goes to. We followed how metadata is fetched in part 2.

In-flight request: a request that has been sent but whose response has not arrived yet. A producer can put several requests on one connection without waiting for responses; the limit is max.in.flight.requests.per.connection (default 5). See section 6.

Idempotent producer: a mode in which the producer attaches a producer id and a sequence number to each batch so that retries do not write the same batch twice. At this commit enable.idempotence defaults to true (ProducerConfig.java lines 574–578), which is why producerId and baseSequence show up in this post's example output. The details are in part 12.

1. What send() does — it stores, it does not send

send(record) calls send(record, null), and send(record, callback) shows the record once to the interceptors (user code plugged in with interceptor.classes) and then goes to doSend() (KafkaProducer.java lines 1147–1151). Most of the producer's work is written out in order in this one method.

Steps 1–12 all happen on the thread that called send(). The Sender thread is only woken up at step 11; sending is done separately by that thread, in section 5.

Let's start after the metadata wait (step 3 in the figure).

            byte[] serializedKey;
            try {
                serializedKey = keySerializerPlugin.get().serialize(record.topic(), record.headers(), record.key());
            } catch (ClassCastException cce) {
                // ...
            }
            byte[] serializedValue;
            // ... the value is serialized the same way ...

            // Try to calculate partition, but note that after this call it can be RecordMetadata.UNKNOWN_PARTITION,
            // which means that the RecordAccumulator would pick a partition using built-in logic (which may
            // take into account broker load, the amount of data produced to each partition, etc.).
            int partition = partition(record, serializedKey, serializedValue, cluster);

            setReadOnly(record.headers());
            Header[] headers = record.headers().toArray();

            int serializedSize = AbstractRecords.estimateSizeInBytesUpperBound(RecordBatch.CURRENT_MAGIC_VALUE,
                    compression.type(), serializedKey, serializedValue, headers);
            ensureValidRecordSize(serializedSize);
            long timestamp = record.timestamp() == null ? nowMs : record.timestamp();

            // Append the record to the accumulator.  Note, that the actual partition may be
            // calculated there and can be accessed via appendCallbacks.topicPartition.
            RecordAccumulator.RecordAppendResult result = accumulator.append(record.topic(), partition, timestamp, serializedKey,
                    serializedValue, headers, appendCallbacks, remainingWaitMs, nowMs, cluster);
            // ...

            if (result.batchIsFull || result.newBatchCreated) {
                log.trace("Waking up the sender since topic {} partition {} is either full or getting a new batch", record.topic(), appendCallbacks.getPartition());
                this.sender.wakeup();
            }
            return result.future;

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java lines 1203–1251

Read in order:

1. Serialization. The key.serializer and value.serializer you configured are called here. From this point the record is byte[], and the producer no longer knows the Java types of key and value. A type mismatch turns the ClassCastException into a SerializationException.
2. Picking a partition. partition() decides a number, or returns "not decided yet" (UNKNOWN_PARTITION) (section 2).
3. Size check. ensureValidRecordSize() throws RecordTooLargeException if the maximum size of this one record exceeds max.request.size (default 1MB) or buffer.memory (lines 1368–1378). This fails before the broker ever sees anything.
4. Timestamp. If the record has no timestamp, the current time (nowMs) is used. This is the CreateTime in the batch header we will see in part 6.
5. Into the waiting room. accumulator.append() appends the record to a batch in that partition's queue (section 3).
6. Waking the Sender. sender.wakeup() is called only when a batch became full or a new batch was opened, not for every record. Adding one record to an existing batch gives the Sender no reason to wake up (that batch will be sent when linger.ms passes anyway).
7. A Future is returned. Nothing has been sent yet at this point. To know whether the record was written to disk you have to wait with get() or receive a callback.

If there is no metadata for the topic, waitOnMetadata() requests an update, wakes the Sender and waits; we followed that path in part 2. What matters here are the two lines that subtract the waiting time from the time budget.

            nowMs += clusterAndWaitTime.waitedOnMetadataMs;
            long remainingWaitMs = Math.max(0, maxBlockTimeMs - clusterAndWaitTime.waitedOnMetadataMs);

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java lines 1200–1201

maxBlockTimeMs is max.block.ms (default 60000, ProducerConfig.java lines 482–487). What is left after waiting for metadata (remainingWaitMs) is passed straight to append() and used for waiting on batch memory (section 8). In other words, max.block.ms is the total time one call to send() may hold the application thread.

Easy to get wrong: send() is "asynchronous", but it can block

Because send() returns a Future, it looks like an asynchronous method that returns immediately. That is true in the sense that it does not wait for a network round trip. But there are two places in the code above where the application thread can stop: when the topic's metadata is not there yet (waitOnMetadata()) and when batch memory runs short (BufferPool.allocate(), section 8). Together they are capped by max.block.ms. Serialization also runs on the application thread, so a heavy serializer slows send() down by that much.

2. Picking a partition — hash if there is a key, "stick" if there isn't

partition() has four branches.

    private int partition(ProducerRecord<K, V> record, byte[] serializedKey, byte[] serializedValue, Cluster cluster) {
        if (record.partition() != null)
            return record.partition();

        if (partitionerPlugin.get() != null) {
            int customPartition = partitionerPlugin.get().partition(
                record.topic(), record.key(), serializedKey, record.value(), serializedValue, cluster);
            // ... IllegalArgumentException if negative ...
            return customPartition;
        }

        if (serializedKey != null && !partitionerIgnoreKeys) {
            // hash the keyBytes to choose a partition
            return BuiltInPartitioner.partitionForKey(serializedKey, cluster.partitionsForTopic(record.topic()).size());
        } else {
            return RecordMetadata.UNKNOWN_PARTITION;
        }
    }

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java lines 1681–1701

Green boxes are cases where the thread that called send() decides the number right away in partition(); yellow boxes are the case where the number is decided inside RecordAccumulator.append().

With a key, partitionForKey() decides: the murmur2 hash of the key bytes, made positive, modulo the partition count.

    public static int partitionForKey(final byte[] serializedKey, final int numPartitions) {
        return Utils.toPositive(Utils.murmur2(serializedKey)) % numPartitions;
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java lines 413–415

So the same key always goes to the same partition as long as the partition count does not change. Kafka's promise that "records with the same key stay in order" starts with this one line.

Without a key: the sticky partitioner

Without a key, partition() returns -1 and the partition is decided inside RecordAccumulator.append(). The BuiltInPartitioner, one per topic, holds "the partition it currently sticks to" and hands out that number (peekCurrentPartitionInfo(), lines 174–186). Instead of rotating partitions record by record, it sticks to one partition and moves only after it has sent enough. The condition for moving is in updatePartitionInfo(), called right after a record is appended.

        if (producedBytes >= stickyBatchSize && enableSwitch || producedBytes >= stickyBatchSize * 2) {
            // We've produced enough to this partition, switch to next.
            StickyPartitionInfo newPartitionInfo = new StickyPartitionInfo(nextPartition(cluster));
            stickyPartitionInfo.set(newPartitionInfo);
        }

clients/src/main/java/org/apache/kafka/clients/producer/internals/BuiltInPartitioner.java lines 254–258

stickyBatchSize is batch.size (the RecordAccumulator passes batchSize when it creates the partitioner in topicInfoFor(), RecordAccumulator.java lines 371–374). enableSwitch is "is the last batch in that partition's queue closed (allBatchesFull())". Put together: move once at least batch.size bytes have gone to the partition and its batches are full; even if the batch is not full, move once it passes twice batch.size. The long comment just above, lines 229–248, explains why: switching partitions with a half-full batch leaves more half-full batches behind and makes batching less efficient.

Why do this at all? If keyless records rotate through the partitions one at a time, as many batches as there are partitions are open at once, and each goes out small when its linger.ms runs out. Sticking to one partition fills one batch quickly so it goes out big. The class comment calls this "adaptive sticky partitioning" and points to KIP-794 (lines 33–39). nextPartition() picks the next partition at random but, true to the word "adaptive", gives a higher probability to partitions whose broker has fewer batches piled up (partitioner.adaptive.partitioning.enable, default true). This post does not follow that weighting (updatePartitionLoadStats()).

Example 1-2: batches of keyless records and keyed records

We sent 600 values of 100 bytes to two three-partition topics with linger.ms=1000 and batch.size=1024: to one without keys, to the other cycling through ten keys key-0 … key-9. This example was run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built unchanged from the base commit (ab1e816411). The example folder is ex/u05-batching; the sending code (Batching.java) and the first experiment are in section 4. Here is the lower part of the output.

=== 2. 3 partitions: records without a key vs with a key (linger.ms=1000 batch.size=1024)
u05-sticky-0: 26 batches, 208 records, records per batch: 8x26
u05-sticky-1: 27 batches, 216 records, records per batch: 8x27
u05-sticky-2: 22 batches, 176 records, records per batch: 8x22
u05-keyed-0: 23 batches, 180 records, records per batch: 8x22 4x1
u05-keyed-1: 23 batches, 180 records, records per batch: 8x22 4x1
u05-keyed-2: 30 batches, 240 records, records per batch: 8x30

8x26 means "26 batches of 8 records" (run.sh counts the count: fields in kafka-dump-log.sh output).

• With keys, each partition's share is fixed. The ten keys hash 3, 3 and 4 to partitions 0, 1 and 2, so the 600 records split 180/180/240. It is the same every run. Partitions 0 and 1 end with one 4-record batch each.
• Without keys, the shares change every run. This time 208/216/176; an earlier run gave 248/96/256. The next partition is chosen at random. But every batch had 8 records, as the condition above says: the producer moves only after the batch on the current partition is full.
• Why 8 is the limit is explained in section 3 (one batch must not exceed 1024 bytes).

Easy to get wrong

• An empty string is a key, not "no key". partition() only checks serializedKey != null. An empty string serialized with StringSerializer becomes a zero-length array, so every record gets the same hash and goes to the same partition.
• Plugging in your own partitioner with partitioner.class turns off both the sticky partitioner and the "adaptive" calculation. The constructor enables adaptive only when partitionerPlugin.get() == null (KafkaProducer.java lines 456–457). To use the sticky behavior even for keyed records instead of hashing, set partitioner.ignore.keys=true (default false, ProducerConfig.java line 443).

3. RecordAccumulator — a queue of batches per partition

Yellow boxes are memory. Only the last batch in a queue is open; all batches in front of it are closed and waiting their turn.

The RecordAccumulator holds a topic → TopicInfo map (topicInfoMap), and a TopicInfo holds a partition → Deque<ProducerBatch> map and the topic's BuiltInPartitioner. Several application threads can touch one queue at the same time (and the Sender thread takes batches out of it), so each queue is locked with synchronized (dq). The lock covers one partition's queue, not the whole topic, so threads writing to different partitions never wait for each other. This is the core of append().

                // check if we have an in-progress batch
                TopicPartition tp = new TopicPartition(topic, effectivePartition);
                Deque<ProducerBatch> dq = topicInfo.batches.computeIfAbsent(tp, k -> new ArrayDeque<>());
                synchronized (dq) {
                    // After taking the lock, validate that the partition hasn't changed and retry.
                    if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
                        continue;

                    RecordAppendResult appendResult = tryAppend(timestamp, key, value, headers, callbacks, dq, nowMs);
                    if (appendResult.appended())
                        return updatePartitionInfoOnAppend(appendResult, topicInfo, partitionInfo, dq, cluster);
                }

                if (buffer == null) {
                    int size = Math.max(this.batchSize, AbstractRecords.estimateSizeInBytesUpperBound(
                            RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), key, value, headers));
                    log.trace("Allocating a new {} byte message buffer for topic {} partition {} with remaining timeout {}ms", size, topic, effectivePartition, maxTimeToBlock);
                    // This call may block if we exhausted buffer space.
                    buffer = free.allocate(size, maxTimeToBlock);
                    // ...
                    nowMs = time.milliseconds();
                }

                synchronized (dq) {
                    // After taking the lock, validate that the partition hasn't changed and retry.
                    if (partitionChanged(topic, topicInfo, partitionInfo, dq, nowMs, cluster))
                        continue;

                    final ByteBuffer batchBuffer = buffer;
                    RecordAppendResult appendResult = appendNewBatch(tp, dq, timestamp, key, value, headers, callbacks,
                            () -> MemoryRecords.builder(batchBuffer, RecordBatch.CURRENT_MAGIC_VALUE, compression, TimestampType.CREATE_TIME, 0L),
                            nowMs);
                    // Set buffer to null, so that deallocate doesn't return it back to free pool, since it's used in the batch.
                    if (appendResult.newBatchCreated)
                        buffer = null;
                    return updatePartitionInfoOnAppend(appendResult, topicInfo, partitionInfo, dq, cluster);
                }

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java lines 322–360

All of this sits inside a while (true) (line 304). How to read it:

1. Lock and try the last batch (tryAppend()). If the record fits, done. Most send() calls end here.
2. If it doesn't fit, release the lock and get memory. free.allocate() waits when memory is short (section 8). Calling it while holding the lock would stop every other thread writing to the same partition, and also the Sender trying to take batches out of that queue. So it is called outside the lock. The size is the larger of batch.size and the maximum size of this one record — a record bigger than batch.size can still sit alone in a batch.
3. Lock again and create a new batch (appendNewBatch()). While the lock was released another thread may already have created a new batch, so appendNewBatch() first tries tryAppend() once more (line 481). If it still doesn't fit, it builds a ProducerBatch on the buffer it got and appends it to the end of the queue (dq.addLast(batch), line 495).
4. Return the buffer if it wasn't used. If step 3 found room in another thread's batch, the buffer is left over; free.deallocate(buffer) in the finally (line 363) gives it back to the pool. If it went into a new batch, buffer = null keeps it from being returned.
5. partitionChanged() and continue. For keyless records the sticky partition was read before locking, so once the lock is acquired another thread may already have moved to another partition. Then everything starts over.

tryAppend() tries the last batch and, if the record doesn't fit, closes that batch.

        ProducerBatch last = deque.peekLast();
        if (last != null) {
            int initialBytes = last.estimatedSizeInBytes();
            FutureRecordMetadata future = last.tryAppend(timestamp, key, value, headers, callback, nowMs);
            if (future == null) {
                last.closeForRecordAppends();
            } else {
                int appendedBytes = last.estimatedSizeInBytes() - initialBytes;
                return RecordAppendResult.appended(future, deque.size() > 1 || last.isFull(), false, appendedBytes);
            }
        }
        return RecordAppendResult.NEEDS_NEW_BATCH;

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java lines 536–547

ProducerBatch.tryAppend() returns null if MemoryRecordsBuilder.hasRoomFor() is false; otherwise it writes the record into the buffer and puts that record's FutureRecordMetadata and callback into the thunks list (ProducerBatch.java lines 168–187). The Future the application received watches the result of this one batch (produceFuture). That is why, when a batch completes, all the record Futures inside it complete together (section 5).

What "full" means for a batch

Whether there is room is decided by MemoryRecordsBuilder. This class writes records straight into the buffer as v2-format bytes: it leaves room for the batch header, writes the records one after another, and fills in the header in close() right before sending (lines 370–395).

    public boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) {
        if (isFull())
            return false;

        // We always allow at least one record to be appended (the ByteBufferOutputStream will grow as needed)
        if (numRecords == 0)
            return true;

        final int recordSize;
        if (magic < RecordBatch.MAGIC_VALUE_V2) {
            recordSize = Records.LOG_OVERHEAD + LegacyRecord.recordSize(magic, key, value);
        } else {
            int nextOffsetDelta = lastOffset == null ? 0 : (int) (lastOffset - baseOffset + 1);
            long timestampDelta = baseTimestamp == null ? 0 : timestamp - baseTimestamp;
            recordSize = DefaultRecord.sizeInBytes(nextOffsetDelta, timestampDelta, key, value, headers);
        }

        // Be conservative and not take compression of the new record into consideration.
        return this.writeLimit >= estimatedBytesWritten() + recordSize;
    }

clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecordsBuilder.java lines 882–901

writeLimit is the remaining size of the buffer it was given (usually batch.size; MemoryRecords.builder() passes buffer.remaining(), MemoryRecords.java lines 584–586). If "bytes written so far + this record" exceeds it, there is no room. The first record is always accepted, whatever its size. Working out why the batches in example 1-2 held 8 records: the batch header is 61 bytes and each record with a 100-byte value is about 109 bytes, so 8 records make 933 bytes and 9 would make 1,042, over 1024 (the size: 933 in example 1-1 is that value).

Easy to get wrong: a "full batch" usually means "a batch that has a successor"

With batch.size=1024, is a batch filled to 933 bytes "full"? Here is isFull().

    public boolean isFull() {
        // note that the write limit is respected only after the first record is added which ensures we can always
        // create non-empty batches (this is used to disable batching when the producer's batch size is set to 0).
        return appendStream == CLOSED_STREAM || (this.numRecords > 0 && this.writeLimit <= estimatedBytesWritten());
    }

clients/src/main/java/org/apache/kafka/common/record/internal/MemoryRecordsBuilder.java lines 919–923

933 < 1024, so false. Records rarely fill batch.size exactly, so a batch usually becomes "full" only when the next record doesn't fit and it is closed (closeForRecordAppends() → CLOSED_STREAM). That is why the readiness condition in section 4 is full = dequeSize > 1 || batch.isFull() — it also checks "there is a next batch in the queue" (line 831). It is also why, in the example in section 4, the last batch waited the full linger.ms.

For reference, this commit includes a ChunkedRecordAccumulator that does not reserve batch.size up front but grows a batch chunk by chunk as records arrive. It is turned on with buffer.memory.allocation.strategy=incremental, but the setting is hidden with defineInternal, the default is full (the existing behavior), and it cannot be combined with compression (KafkaProducer.java lines 467–517, ProducerConfig.java lines 421–428). 4.4.0 has the same class and setting. This post only covers the default, full.

4. When to send — one if in ready()

The Sender thread asks RecordAccumulator.ready() when to send. ready() walks the topics and partitions, looks at the first batch of each queue, and if that batch is worth sending, adds the partition's leader broker to the set of "nodes with something to send". The decision lives in batchReady().

    private long batchReady(boolean exhausted, Node leader,
                            long waitedTimeMs, boolean backingOff, int backoffAttempts,
                            boolean full, long nextReadyCheckDelayMs, Set<Node> readyNodes) {
        long timeToWaitMs = backingOff ? retryBackoff.backoff(backoffAttempts > 0 ? backoffAttempts - 1 : 0) : lingerMs;
        boolean expired = waitedTimeMs >= timeToWaitMs;
        boolean transactionCompleting = transactionManager != null && transactionManager.isCompleting();
        boolean sendable = full
| expired
| exhausted
| closed
| flushInProgress()
| transactionCompleting;
        if (sendable && !backingOff) {
            readyNodes.add(leader);
        } else {
            long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);
            // Note that this results in a conservative estimate since an un-sendable partition may have
            // a leader that will later be found to have sendable data. However, this is good enough
            // since we'll just wake up and then sleep again for the remaining time.
            nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);
        }
        return nextReadyCheckDelayMs;
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java lines 714–736

This is the if where linger.ms is used. The lingerMs field becomes timeToWaitMs, and once the batch has waited (waitedTimeMs, at first the time since it was created) longer than that, it is expired. The six conditions of sendable in a table:

A batch not worth sending yet leaves behind "when to look again" (timeLeftMs). The smallest of these across all partitions becomes nextReadyCheckDelayMs, which lets the Sender sleep only that long (section 5). A batch being retried (backingOff) uses the retry backoff instead of linger.ms (section 7).

To keep each partition queue locked as briefly as possible, ready() only reads a few values of the first batch inside the lock and makes the decision (batchReady()) outside it. A comment says this loop is "especially hot with large partition counts" and warns against adding contention with the threads calling send() (lines 801–808).

Example 1-1: changing linger.ms and batch.size

We sent the same records (200 values of 100 bytes, no key) to single-partition topics with four settings, and counted with kafka-dump-log.sh how many batches ended up on disk. Here is the core of the sending code.

// An example written for this post (not part of Kafka's source)
// Sends the same records with different settings only. run.sh checks the batches with kafka-dump-log.sh.
public class Batching {
    public static void main(String[] args) throws Exception {
        // ... read topic, record count, linger.ms, batch.size, sync/async from args, and set bootstrap.servers and the serializers ...
        props.put(ProducerConfig.LINGER_MS_CONFIG, linger);          // "linger.ms"
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);      // "batch.size"

        String value = "v".repeat(100);                               // values are always 100 bytes
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.partitionsFor(topic);                            // fetch metadata first (to keep it out of the timing)
            CountDownLatch done = new CountDownLatch(count);
            long start = System.nanoTime();
            for (int i = 0; i < count; i++) {
                String key = withKey ? "key-" + (i % 10) : null;     // without a key, the built-in partitioner picks
                ProducerRecord<String, String> rec = new ProducerRecord<>(topic, key, value);
                if (sync) {
                    producer.send(rec).get();                         // the next record only after the response
                    done.countDown();
                } else {
                    producer.send(rec, (md, e) -> done.countDown());  // queue it and move straight on
                }
            }
            // ... print when the send() loop ended and when all callbacks arrived ...
        }
    }
}

[Shortened from the example's ex/u05-batching/Batching.java]

=== 1. 200 records of 100 bytes, four producer settings
u05-sync: linger.ms=0 batch.size=16384 sync -> send() loop 1790 ms, all acked 1790 ms
u05-small: linger.ms=1000 batch.size=1024 async -> send() loop 110 ms, all acked 1119 ms
u05-large: linger.ms=1000 batch.size=16384 async -> send() loop 157 ms, all acked 1181 ms
u05-default: linger.ms=5 batch.size=16384 async -> send() loop 56 ms, all acked 173 ms
--- batches on disk (kafka-dump-log.sh)
u05-sync-0: 200 batches, 200 records, records per batch: 1x200
u05-small-0: 25 batches, 200 records, records per batch: 8x25
u05-large-0: 2 batches, 200 records, records per batch: 148x1 52x1
u05-default-0: 3 batches, 200 records, records per batch: 148x1 43x1 9x1
--- first two batch headers of u05-small
baseOffset: 0 lastOffset: 7 count: 8 baseSequence: 0 lastSequence: 7 producerId: 1 CreateTime: 1794025632545 size: 940
baseOffset: 8 lastOffset: 15 count: 8 baseSequence: 8 lastSequence: 15 producerId: 1 CreateTime: 1794025632555 size: 933

Times (ms), CreateTime and producerId differ from run to run. The batch header lines keep only the fields we need.

The batch shapes the four settings produced. All numbers are the actual output of Example 1-1 above (times and the default setting's batch boundaries vary per run).

1. u05-sync: 200 batches. With send().get() sending the next record only after each response, two records never get the chance to meet in the waiting room. Batching is created not by a producer setting but by the application calling send() repeatedly without waiting. With 200 round trips it is also the slowest.
2. u05-small: 25 batches of 8. 8 records fit in batch.size=1024 (section 3), so each time the 9th record arrives the previous batch is closed (full) and goes out immediately. Yet all callbacks arrived at 1,119 ms. The 25th batch holds 8 records, but isFull() is false and no batch follows it, so it is not full; it went out as expired only after waiting the whole linger.ms=1000.
3. u05-large: 148 and 52. A 16KB batch holds 148 records. The first batch is closed when the 149th record arrives and goes out at once; the 52-record remainder again waited one second. Another run gave 147 and 53. The timestamp difference inside each record (timestampDelta) is written as a variable-length integer (DefaultRecord.java line 541), so when the first records go out slowly and that difference grows, each record gets a byte bigger and one fewer fits (the first header above being size: 940, 7 bytes more than the second batch, looks like the same cause). The order in which run.sh prints 148x1 52x1 is not the batch order.
4. u05-default: with the default (linger.ms=5) the result changes every run. This time 148/43/9; earlier runs gave 148/52 and 83/117. 5 milliseconds is shorter than one pass of the send() loop (56 ms), so batch boundaries depend on when the Sender wakes up and looks at ready(). It is still far faster than 200 round trips (case 1).
5. producerId: 1 and baseSequence: 0/8 in the headers were attached by the idempotent producer (enable.idempotence=true is the default). The sequence continues from batch to batch. We glance at where the number is attached in section 6; what it is for is in part 12.

Easy to get wrong

• linger.ms defaults to 5 both at this commit and in 4.4.0 (ProducerConfig.java line 445). If an older article says "default 0", it describes the release of its time (this post did not check in the source when it changed).
• batch.size=0 means "no batching". The constructor raises it to 1 with Math.max(1, …) (KafkaProducer.java line 466), and hasRoomFor() accepts only the first record, so every batch holds one record.
• linger.ms is not "wait this long, then send" but "send no later than this". If the batch fills first (full) there is no waiting. Conversely, even with linger.ms=0, records that pile up while the Sender is busy with something else go out together in one batch.

5. One turn of the Sender thread

The Sender thread is created at the end of the KafkaProducer constructor.

            this.sender = newSender(logContext, kafkaClient, this.metadata, clientInstanceId);
            String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
            this.ioThread = new Sender.SenderThread(ioThreadName, this.sender, true);
            this.ioThread.start();

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java lines 520–523

NETWORK_THREAD_PREFIX is "kafka-producer-network-thread" (line 255), and the third argument true makes it a daemon thread. One thread per producer. No matter how many brokers or partitions, this single thread handles every connection. Its body is the loop in Sender.run().

        // main loop, runs until close is called
        while (running) {
            try {
                runOnce();
            } catch (Exception e) {
                log.error("Uncaught error in kafka producer I/O thread: ", e);
            }
        }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 244–251

Leaving transactions aside, runOnce() is two lines: send what can be sent (sendProducerData()), then run the network (client.poll()).

        long currentTimeMs = time.milliseconds();
        long pollTimeout = sendProducerData(currentTimeMs);
        client.poll(pollTimeout, currentTimeMs);
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 343–346

Every step happens on the single Sender thread. The response (step 9) usually arrives not in the same turn but in the poll() of some later turn.

The outline of sendProducerData() is steps 1–6 in the figure.

1. accumulator.ready() collects the leader nodes that have something to send, using the decision from section 4. If a partition's leader is unknown, it requests a metadata update.
2. client.ready(node, now) checks whether we can send to that node now. If there is no connection, it starts connecting and drops the node from this turn (NetworkClient.java lines 434–446). It also drops the node if the in-flight limit is reached (section 6).
3. accumulator.drain() walks, for each remaining node, the partitions that node leads and takes one batch from the head of each queue, until the total would exceed max.request.size (default 1MB) (lines 976–1060). Each drained batch is finished with batch.close(), which fills in its header. Always starting from the same partition would starve the later ones, so each node remembers where it stopped (drainIndex) and continues from there next time (the comment on line 982: "to make starvation less likely").
4. Expiry clean-up fails batches whose delivery.timeout.ms has passed (section 7).
5. sendProduceRequests() builds one Produce request per node. The batches' MemoryRecords go as they are into the ProduceRequestData we saw in part 3, per topic and partition (lines 896–949). The application's acks setting becomes the request's Acks field here (setAcks(acks)); ProducerConfig has already turned acks=all into -1 (line 707). With acks=0, newClientRequest(…, acks != 0, …) makes it a request that expects no response, which completes the moment it is fully written to the socket (NetworkClient.java lines 1061–1070). When the broker responds for each acks value is covered in part 6 and part 9.
6. pollTimeout is 0 if there were nodes to send to (go straight to the next turn); otherwise the earlier of nextReadyCheckDelayMs from section 4 (the earliest linger.ms expiry) and the next batch expiry (if step 2 dropped a node, the time until it may try to connect to that node again is also taken into account) (lines 437–452).

Then client.poll(pollTimeout) runs the Selector. This Selector is exactly the org.apache.kafka.common.network.Selector that the broker's Processor used in part 4. Strictly speaking, the Sender sleeps inside it. When an application thread calls sender.wakeup(), client.wakeup() → Selector.wakeup() interrupts that sleep (Sender.java lines 963–965).

        long metadataTimeout = metadataUpdater.maybeUpdate(now);
        long telemetryTimeout = telemetrySender != null ? telemetrySender.maybeUpdate(now) : Integer.MAX_VALUE;
        try {
            this.selector.poll(Utils.min(timeout, metadataTimeout, telemetryTimeout, defaultRequestTimeoutMs));
        } catch (IOException e) {
            log.error("Unexpected error during I/O", e);
        }

        // process completed actions
        long updatedNow = this.time.milliseconds();
        List<ClientResponse> responses = new ArrayList<>();
        handleCompletedSends(responses, updatedNow);
        handleCompletedReceives(responses, updatedNow);
        // ... dropped connections, new connections, ApiVersions, timeouts ...
        completeResponses(responses);

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 718–737

completeResponses() calls the completion handler attached to each request. For a Produce request that handler is Sender.handleProduceResponse() (line 942), which uses the per-partition results to go completeBatch() → ProducerBatch.done() → completing each record's Future and calling the callbacks (ProducerBatch.java lines 318–345). In other words, callbacks are called on the Sender thread.

Example 2: the thread that calls send() and the thread that runs the callback

We created a producer with client.id=u05-app, sent one record, and printed the thread name and call stack (Kafka frames only) inside the callback. Then we sent a 2MB value, larger than max.request.size (1MB). The example folder is ex/u05-threads.

// An example written for this post (not part of Kafka's source)
// Compares the thread that calls send() with the thread the callback runs on, and prints the call stack inside the callback.
            // ... create the producer (client.id=u05-app) and print the threads whose names start with kafka-producer ...
            System.out.println("send() called on: " + Thread.currentThread().getName());
            CountDownLatch done = new CountDownLatch(1);
            producer.send(new ProducerRecord<>("u05-threads", "k", "hello"), (md, e) -> {
                System.out.println("callback ran on:  " + Thread.currentThread().getName()
                        + " (partition " + md.partition() + ", offset " + md.offset() + ")");
                // ... print only the org.apache.kafka frames of new Throwable().getStackTrace(), bottom to top ...
                done.countDown();
            });
            done.await();

            // 3) a record that fails right inside send(): a value larger than max.request.size (1MB by default)
            String big = "x".repeat(2 * 1024 * 1024);
            producer.send(new ProducerRecord<>("u05-threads", "k", big), (md, e) ->
                    System.out.println("callback for a too-large record ran on: " + Thread.currentThread().getName()
                            + " -> " + e.getClass().getSimpleName()));
            System.out.println("send() of the too-large record returned");

[Shortened from the example's ex/u05-threads/Threads.java]

=== which thread runs what
thread after new KafkaProducer(): kafka-producer-network-thread | u05-app daemon=true
send() called on: main
callback ran on:  kafka-producer-network-thread | u05-app (partition 0, offset 0)
callback stack (bottom = first called, Kafka frames only):
  clients.producer.internals.Sender.run:247
  clients.producer.internals.Sender.runOnce:345
  clients.NetworkClient.poll:737
  clients.NetworkClient.completeResponses:745
  clients.ClientResponse.onComplete:154
  clients.producer.internals.Sender.lambda$sendProduceRequest$0:942
  clients.producer.internals.Sender.handleProduceResponse:609
  clients.producer.internals.Sender.lambda$handleProduceResponse$0:609
  clients.producer.internals.Sender.lambda$handleProduceResponse$1:635
  clients.producer.internals.Sender.completeBatch:733
  clients.producer.internals.Sender.completeBatch:762
  clients.producer.internals.ProducerBatch.complete:243
  clients.producer.internals.ProducerBatch.done:297
  clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks:333
  clients.producer.KafkaProducer$AppendCallbacks.onCompletion:1807
  (user callback)
callback for a too-large record ran on: main -> RecordTooLargeException
send() of the too-large record returned

1. Just creating a producer starts one thread. Its name ends with client.id, and it is a daemon thread. Without a client.id it gets a numbered name like producer-1 (see example 3's output).
2. The callback runs on the Sender thread, not on main. Read from bottom to top, the stack is steps 7–12 of the figure above: Sender.run → client.poll() in runOnce (line 345) → NetworkClient.completeResponses → handleProduceResponse → completeBatch → ProducerBatch.done → completeFutureAndFireCallbacks → the application's callback.
3. But the callback for the too-large record ran on main. ensureValidRecordSize() from section 1 throws RecordTooLargeException inside send(), and doSend()'s catch (ApiException e) calls the callback on the spot and returns a failed Future (lines 1252–1267). send() returned without throwing.

Easy to get wrong: slow work in a callback stalls the whole producer

A callback running on the Sender thread means that until the callback returns, all sending and response handling of that producer stops. There is only one Sender, and callbacks are called one after another inside client.poll(). If a callback writes to a database or waits on another future.get(), no batches go out in the meantime, and once the waiting room fills up even send() blocks (section 8). Hand heavy work to another thread (an executor). And as case 3 of example 2 shows, a callback can also run on the thread that called send(). Code that assumes callbacks always run on the same thread can be wrong.

6. In-flight requests and ordering — two kinds of "mute"

One of the reasons client.ready(node) returns false is the in-flight limit. NetworkClient queues the requests sent to each node in InFlightRequests and removes them as responses arrive.

    public boolean canSendMore(String node) {
        Deque<NetworkClient.InFlightRequest> queue = requests.get(node);
        return queue == null || queue.isEmpty() ||
               (queue.peekFirst().send.completed() && queue.size() < this.maxInFlightRequestsPerConnection);
    }

clients/src/main/java/org/apache/kafka/clients/InFlightRequests.java lines 96–100

There are two conditions: the most recently sent request has been fully written to the socket (send.completed() — new requests go to the front of this queue), and there are fewer in-flight requests than max.in.flight.requests.per.connection. With the default of 5 (ProducerConfig.java lines 520–525), up to five Produce requests go to one broker without waiting for responses, so the connection does not sit idle during the round trip.

In part 4 we saw that the broker processes the requests of one connection one at a time (the Processor mutes the connection after reading a request). The two do not clash. Even with five requests sent ahead, the broker reads what is in the socket buffer one by one and processes them in order, and responses come back in the order they were sent. The client limit is about "how many to keep on the wire"; the broker's mute is about "how many to process at once".

Another mute on the producer side

With max.in.flight.requests.per.connection=1, the producer turns on one more mechanism. newSender() passes maxInflightRequests == 1 as guaranteeMessageOrder (KafkaProducer.java line 599), and the Sender mutes the partition of every batch it drains.

        Map<Integer, List<ProducerBatch>> batches = this.accumulator.drain(metadataSnapshot, result.readyNodes, this.maxRequestSize, now);
        addToInflightBatches(batches);
        if (guaranteeMessageOrder) {
            // Mute all the partitions drained
            for (List<ProducerBatch> batchList : batches.values()) {
                for (ProducerBatch batch : batchList)
                    this.accumulator.mutePartition(batch.topicPartition);
            }
        }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 418–426

ready() and drain() skip muted partitions (isMuted(tp), line 992), and the mute is lifted at the end of completeBatch() after that batch's response is handled (lines 737–738). Keeping only one batch per partition on the wire prevents a later batch from being written first while an earlier one fails and is retried, which would swap their order. The word is the same, "mute", but in part 4 the broker pauses a connection, while here the producer pauses a partition.

Two things called "mute". The broker pauses a connection; the producer (with in-flight 1) pauses a partition. The client limit is about how many to keep on the wire.

The idempotent producer (the default) keeps order without this mechanism. When drain() takes a batch, it attaches the producer id and sequence number (batch.setProducerState(…), line 1042), and the broker checks the numbers for duplicates and gaps. In exchange, ProducerConfig raises a configuration error if idempotence is on and max.in.flight.requests.per.connection is greater than 5 (lines 672–676). The detailed rules for keeping order with sequence numbers are in part 12.

7. Retries and delivery.timeout.ms — the life of a batch

The state names are for explanation. The source's ProducerBatch.FinalState has only the three final states (ABORTED, FAILED, SUCCEEDED) (line 64). ABORTED, the end for a forced close or an aborted transaction, is left out of the figure.

When a response carries an error, completeBatch() decides whether to retry.

        } else if (error != Errors.NONE) {
            if (canRetry(batch, response, now)) {
                log.warn(
                    "Got error produce response with correlation id {} on topic-partition {}, retrying ({} attempts left). Error: {}",
                    correlationId,
                    batch.topicPartition,
                    this.retries - batch.attempts() - 1,
                    formatErrMsg(response));
                reenqueueBatch(batch, now);
            } else if (error == Errors.DUPLICATE_SEQUENCE_NUMBER) {
                // ...
                completeBatch(batch, response);
            } else {
                // ...
                failBatch(batch, response, batch.attempts() < this.retries, true);
            }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 691–712

    private boolean canRetry(ProducerBatch batch, ProduceResponse.PartitionResponse response, long now) {
        return !batch.hasReachedDeliveryTimeout(accumulator.getDeliveryTimeoutMs(), now) &&
            batch.attempts() < this.retries &&
            !batch.isDone() &&
            (transactionManager == null ?
                    response.error.exception() instanceof RetriableException :
                    transactionManager.canRetry(response, batch));
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java lines 876–883

Four conditions for a retry: delivery.timeout.ms has not passed since the batch was created, the attempt count is below retries, the batch is not done yet, and the error is of a retriable kind (a subclass of RetriableException — NotLeaderOrFollowerException when the leader moved, NotEnoughReplicasException from part 9, and so on). For idempotent and transactional producers, the last decision is left to the TransactionManager.

A batch to be resent is put back at the front of its partition queue by RecordAccumulator.reenqueue() (lines 603–612), since it must go out before the batches queued behind it. A producer with idempotence turned off simply does deque.addFirst(batch); the default idempotent producer (which has a TransactionManager) uses insertInSequenceOrder() to slot it in where the sequence numbers stay in order. That is usually the front, but when several batches of the same partition are retried together, it goes behind the ones with lower numbers (lines 659–699). Its attempts goes up by one, which makes backingOff from section 4 true, so instead of linger.ms it waits a backoff that grows from retry.backoff.ms (default 100) up to retry.backoff.max.ms (default 1000) (shouldBackoff(), lines 920–936). A retry that knows the leader has changed does not wait, though (!hasLeaderChanged).

The real limit is delivery.timeout.ms, not retries

retries defaults to Integer.MAX_VALUE (ProducerConfig.java line 429). There is effectively no count limit; what ends retrying is time. delivery.timeout.ms (default 120000, line 446) is the upper bound on "from send() returning until the result — success or failure — is reported". It is measured from when the batch was created (createdMs).

    boolean hasReachedDeliveryTimeout(long deliveryTimeoutMs, long now) {
        return deliveryTimeoutMs <= now - this.createdMs;
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java lines 462–464

On every turn the Sender looks for batches past this time in two places: from the head of each queue in the waiting room (accumulator.expiredBatches(), lines 572–593), and among batches sent and awaiting a response (getExpiredInflightBatches(), lines 185–216). Those it finds fail with a TimeoutException reading "Expiring N record(s) for … ms has passed since batch creation" (lines 368–373). Even if the broker never answers, the application gets a result within this time.

The constructor also checks that this value is not too small. If delivery.timeout.ms is less than linger.ms + request.timeout.ms, it throws ConfigException when you set the value yourself, and when it is the default it raises it to that sum and only logs a warning (KafkaProducer.java lines 637–657). There must be room to wait linger.ms once and to send one request and wait for its response. When we experimented with a 60-second linger.ms for example 3, we hit this check and had to add delivery.timeout.ms=90000.

Easy to get wrong

• A record does not fail just because request.timeout.ms passed. When a request times out, NetworkClient closes the connection to that node (handleTimedOutRequests(), NetworkClient.java lines 1019–1027), and handleProduceResponse() passes the batch to completeBatch() with a REQUEST_TIMED_OUT error (Sender.java lines 583–588). That error's exception (TimeoutException) is retriable, so if delivery.timeout.ms has time left, the batch is sent again. The application only sees a failure when delivery.timeout.ms has passed or the error is not retriable.
• delivery.timeout.ms is measured from when the batch was created, not from when you called send(). A record that joined an already-open batch later gets correspondingly less time.

8. BufferPool — buffer.memory and max.block.ms

Batch buffers are handed out by the BufferPool. The pool's total size is buffer.memory (default 32MB, ProducerConfig.java line 420), and buffers of exactly batch.size are not thrown away after use but kept in a free list for reuse, so that a new 16KB ByteBuffer does not have to be created and left to the GC every time (class comment, lines 38–46).

            // check if we have a free buffer of the right size pooled
            if (size == poolableSize && !this.free.isEmpty())
                return this.free.pollFirst();

            // now check if the request is immediately satisfiable with the
            // memory on hand or if we need to block
            int freeListSize = freeSize() * this.poolableSize;
            if (this.nonPooledAvailableMemory + freeListSize >= size) {
                // we have enough unallocated or pooled memory to immediately
                // satisfy the request, but need to allocate the buffer
                freeUp(size);
                this.nonPooledAvailableMemory -= size;
            } else {
                // we are out of memory and will have to block
                int accumulated = 0;
                Condition moreMemory = this.lock.newCondition();
                // ... put moreMemory at the end of waiters and await() repeatedly until enough is gathered ...

clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java lines 155–170

allocate() is called by application threads, deallocate() mostly by the Sender thread. They are connected by one ReentrantLock and a Condition per waiter.

When memory is short, the application thread puts its own Condition at the end of the waiters queue and waits. The queue is there for fairness: the class comment says "all memory is given to the longest waiting thread until it has sufficient memory", so that a thread waiting for a large chunk is not starved forever by small requests. The upper bound of the wait is remainingWaitMs from section 1 (= max.block.ms minus the time spent waiting for metadata); past it comes BufferExhaustedException.

Memory goes back through deallocate(). When a batch ends, successfully or not, the Sender returns its buffer: to the free list if it is batch.size, otherwise back into the free-memory counter, and then it wakes the first waiter.

    public void deallocate(ByteBuffer buffer, int size) {
        lock.lock();
        try {
            if (size == this.poolableSize && size == buffer.capacity()) {
                buffer.clear();
                this.free.add(buffer);
            } else {
                this.nonPooledAvailableMemory += size;
            }
            Condition moreMem = this.waiters.peekFirst();
            if (moreMem != null)
                moreMem.signal();
        } finally {
            lock.unlock();
        }
    }

clients/src/main/java/org/apache/kafka/clients/producer/internals/BufferPool.java lines 481–496

So batch memory is only freed when the response arrives. Sending does not free it; it comes back only when the broker responds and the batch ends. If the broker is slow, the waiting room and the in-flight batches pile up holding memory, and eventually send() blocks.

Example 3: buffer.memory set to two batches

With batch.size=16384, buffer.memory=32768 (two batches) and max.block.ms=3000, we sent one record to each of three partitions. Each partition needs its own batch, so the third send() must wait for one of the earlier batches to end and return its memory. In the second experiment we stopped the broker process with kill -STOP just before sending (the connection stays alive but no response comes). The example folder is ex/u05-buffer-memory.

// An example written for this post (not part of Kafka's source)
// Shrinks buffer.memory to two batches (32KB) and sends one record to each of three partitions.
        // ... read the topic from args, and set bootstrap.servers and the serializers ...
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");          // one batch = 16KB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "32768");       // total = two batches
        props.put(ProducerConfig.LINGER_MS_CONFIG, args[1]);
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "3000");         // how long send() may wait for memory
        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, "90000"); // must be at least linger.ms + request.timeout.ms

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // fetch metadata and open the connection to the leader in advance (flush() sends without waiting for linger)
            producer.send(new ProducerRecord<>(topic, 0, null, "warm-up"));
            producer.flush();
            // ... (second experiment) kill -STOP the broker here ...
            for (int p = 0; p < 3; p++) {
                // ... send one record to partition p and print how long send() took and the callback ...
            }
            producer.close(Duration.ofSeconds(5));
        }

[Shortened from the example's ex/u05-buffer-memory/BufferMemory.java]

=== 1. buffer.memory=32768 batch.size=16384 max.block.ms=3000, linger.ms=5
send() to partition 0 returned after 16 ms
send() to partition 1 returned after 0 ms
  callback p0 at   31 ms on kafka-producer-network-thread | producer-1: offset 1
send() to partition 2 returned after 28 ms
  callback p1 at   50 ms on kafka-producer-network-thread | producer-1: offset 0
  callback p2 at   65 ms on kafka-producer-network-thread | producer-1: offset 0
=== 2. the same, but the broker is paused before sending
broker paused (kill -STOP)
send() to partition 0 returned after 9 ms
send() to partition 1 returned after 0 ms
  callback p2 at 3028 ms on main: BufferExhaustedException: Failed to allocate 16384 bytes within the configured max blocking time 3000 ms. Total memory: 32768 bytes. Available memory: 0 bytes. Poolable size: 16384 bytes
send() to partition 2 returned after 3017 ms
  callback p1 at 8043 ms on kafka-producer-network-thread | producer-1: KafkaException: Producer is closed forcefully.
  callback p0 at 8044 ms on kafka-producer-network-thread | producer-1: KafkaException: Producer is closed forcefully.

Times (ms) vary a little from run to run, and so does the order in which the p0 and p1 callbacks print.

1. Experiment 1: the third send() took 28 ms. The first two send() calls returned at once, but the third waited for memory; meanwhile the earlier batches went out after linger.ms (5 ms), and in this run the p0 response came back first, at 31 ms. That deallocate() woke the third send() (which is why callback p0 is printed before the third send() returns). In another run the p1 response arrived first and p1 appeared in the same place.
2. Experiment 2: no response, no memory back. The third send() blocked for exactly max.block.ms (3 seconds) and ended with BufferExhaustedException, whose message is the sentence inside the waiting loop that the allocate() excerpt above cut (BufferPool.java lines 178–180). Like the too-large record in example 2, this exception is delivered to the callback on main, and send() returned without throwing (BufferExhaustedException → TimeoutException → RetriableException → ApiException, so doSend()'s catch (ApiException e) catches it).
3. p0 and p1 failed around 8 seconds, when close(5 seconds) ran out of time and closed forcefully.

Easy to get wrong: a thread waiting for memory does not immediately wake the Sender

The exhausted condition in the table of section 4 (if a thread is waiting for memory, send even non-full batches) suggests the producer will push batches out by itself when memory is short. But neither BufferPool nor RecordAccumulator contains a call that wakes the Sender (wakeup) — we searched both files. exhausted is only used when the Sender wakes up for some other reason and looks at ready(). In fact, running experiment 1 with linger.ms=60000 gave different results from run to run: sometimes the third send() failed with BufferExhaustedException after 3 seconds (the Sender was in a 60-second sleep), sometimes it succeeded within tens of milliseconds (apparently the Sender happened to wake up for some other network work, saw exhausted and sent the batches — we could not confirm what woke it). Because the result was not stable, we left it out of the post's examples. If you use a large linger.ms, it is safer to leave plenty of buffer.memory.

Settings at a glance

The path from the setting names you use in applications to the source. All constants are in ProducerConfig, and the defaults were read from this commit's define(…) calls.

What this post leaves out

• Idempotence and transactions (TransactionManager, InitProducerId, sequence checks): clients/src/main/java/org/apache/kafka/clients/producer/internals/TransactionManager.java — part 12
• How metadata is fetched (waitOnMetadata(), ProducerMetadata): from KafkaProducer.java line 1299 — part 2
• The weighting of adaptive partitioning (updatePartitionLoadStats(), partitioner.availability.timeout.ms): from BuiltInPartitioner.java line 273
• Splitting and resending a batch that is too large (MESSAGE_TOO_LARGE → splitAndReenqueue()): from RecordAccumulator.java line 618
• Compression (compression.type) and compression-ratio estimation: estimatedCompressionRatio in MemoryRecordsBuilder; how the broker handles compressed batches is in part 6
• The chunk-by-chunk memory strategy (ChunkedRecordAccumulator, BufferPool.allocateChunks()): clients/src/main/java/org/apache/kafka/clients/producer/internals/ChunkedRecordAccumulator.java
• How flush() and close() send the remaining batches: Sender.java lines 253–262, RecordAccumulator.beginFlush()·awaitFlushCompletion()
• Connecting, ApiVersions negotiation and reconnect backoff: NetworkClient — negotiation is in part 3
• Client telemetry (ClientTelemetryReporter), interceptors (ProducerInterceptors), security (SASL, TLS)

Summary

What's next

We have followed the Sender to the point where it packs batches into a Produce request and sends it to a broker. That request passes the network layer of part 4 and reaches KafkaApis.handleProduceRequest(). In the next post we follow how the broker checks those batches, assigns offsets and appends them to the log (ReplicaManager → Partition → UnifiedLog), and take apart, field by field, the record-batch bytes that MemoryRecordsBuilder wrote in this post. There the design of the broker writing the producer's batch to disk almost unchanged becomes visible. Continue with part 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path.

What the broker waits for before answering when acks=all is in part 9, and what the producer id and sequence numbers we kept postponing are for is in part 12. One thread handling many connections with an NIO Selector also resembles the Poller in Tomcat part 5.

• Part 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Part 3: The Wire Protocol — Requests and Responses Generated from JSON Schemas
• Part 4: The Broker's Network Layer — Acceptors, Processors and Request Handlers
• Part 11: The Consumer and the Group Coordinator — Inside poll() and Rebalancing