Digging into Kafka's Source 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path

What this post covers

This is the sixth post in the "Digging into Kafka's Source" series. In Part 4 we followed a request through the network thread (the Processor) until it reached KafkaApis.handle() on a request handler thread, and in Part 5 we saw the producer gather the records passed to send() into batches and the Sender thread ship them as Produce requests. This time we go where those two meet: from the moment the broker receives a Produce request until the records land in a log file on disk.

We want to answer four questions.

• What does the "record batch" sent by the producer look like in bytes? Where inside it are the keys, values, headers and timestamps you set in your app?
• Which classes and methods does one Produce request pass through inside the broker before it reaches a file?
• Who assigns offsets, and when? And why does the broker write the batch almost as it came, instead of unpacking and rebuilding it?
• Where in the broker's code do acks=0, acks=1 and acks=all change when the response is sent?

All line numbers in this post refer to this commit. Part 1 shows how to get the same commit and build the Docker image used for the examples. The write path read here lives in the same files, in the same shape, in 4.4.0, the latest stable release most people use. I checked on the 4.4.0 tag that KafkaApis.handleProduceRequest() calls replicaManager.handleProduceAppend(), and that the min.insync.replicas check in Partition, the in-place conditions in LogValidator, the "one batch per partition" check and the producer's acks default of all are all the same.

UnifiedLog.java is 2,806 lines and ReplicaManager.scala is 2,577. Even a single write is tangled up with idempotent-producer and transaction checks, quotas, segment rolling and waiting for replication. This post picks only the main path a newcomer can follow; whatever was left out is listed by name and location in "What this post leaves out" at the end.

The map from Part 2. This post zooms into the write path that runs from C (request processing) through D (partitions) down to E (log storage). The inside of E (segments, indexes) belongs to Part 7, and replication in D to Part 9.

Words to know first

Record batch: a bundle of several records headed for the same partition. The producer sends in batches, the broker writes to disk in batches, and the consumer receives batches too. Sections 1–2 of this post are about the bytes of this bundle. How the producer collects batches was Part 5.

magic: the version number of the record format. The current format is magic 2 ("v2" in this post); the old v0 and v1 formats are not accepted in Produce requests v3 and later (Section 3). This is a different number from the version of the Produce request itself (v3–v13), so keep them apart.

Offset and log end offset: an offset is the number saying which position a record holds within a partition. It starts at 0, grows by one, and never changes once assigned. The log end offset (LEO) is "the offset the next record will get", i.e. the last record's offset + 1. As Section 4 shows, the broker assigns offsets starting from this value.

Leader, ISR, min.insync.replicas: every partition has several copies (replicas), and one of them is the leader, which takes the writes (Part 2). The ISR (in-sync replicas) is the list of replicas that are keeping up with the leader (the leader included). The topic setting min.insync.replicas says "how large the ISR must at least be to accept an acks=all write" (default 1, ServerLogConfigs.java line 159).

High watermark (HW): the offset up to which every replica in the ISR has the data. Consumers can only read up to the high watermark. In this post we only go as far as "acks=all waits for this value to catch up"; how it moves is Part 9.

Leader epoch: a number that grows each time the partition's leader changes. When the broker writes a batch, it records in the batch header (partitionLeaderEpoch) "under which leader this batch was written". Replicas use it to reconcile their logs after a leader change; that use is Part 9's topic.

varint: a variable-length integer format that writes small numbers in few bytes. Section 2 covers it in detail.

Purgatory: a waiting room where requests that cannot finish right away sit until their condition is met. An acks=all Produce request waits for replication here. Its structure is covered in Part 8 and Part 9.

1. Record batch v2 — a 61-byte header

Field order

The batch format is spelled out in the class comment of DefaultRecordBatch.

 * RecordBatch =>
 *  BaseOffset => Int64
 *  Length => Int32
 *  PartitionLeaderEpoch => Int32
 *  Magic => Int8
 *  CRC => Uint32
 *  Attributes => Int16
 *  LastOffsetDelta => Int32 // also serves as LastSequenceDelta
 *  BaseTimestamp => Int64
 *  MaxTimestamp => Int64
 *  ProducerId => Int64
 *  ProducerEpoch => Int16
 *  BaseSequence => Int32
 *  RecordsCount => Int32
 *  Records => [Record]

clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java lines 48–62

This order is not only in a comment: the constants right below compute each byte position as "the previous field's position + the previous field's length", one field at a time.

    static final int BASE_OFFSET_OFFSET = 0;
    static final int BASE_OFFSET_LENGTH = 8;
    static final int LENGTH_OFFSET = BASE_OFFSET_OFFSET + BASE_OFFSET_LENGTH;
    static final int LENGTH_LENGTH = 4;
    static final int PARTITION_LEADER_EPOCH_OFFSET = LENGTH_OFFSET + LENGTH_LENGTH;
    static final int PARTITION_LEADER_EPOCH_LENGTH = 4;
    static final int MAGIC_OFFSET = PARTITION_LEADER_EPOCH_OFFSET + PARTITION_LEADER_EPOCH_LENGTH;
    static final int MAGIC_LENGTH = 1;
    public static final int CRC_OFFSET = MAGIC_OFFSET + MAGIC_LENGTH;
    // ... (the same pattern up to BASE_SEQUENCE)

clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java lines 104–112

At the end of that chain, RECORDS_OFFSET (line 130), where the first record starts, is 8+4+4+1+4+2+4+8+8+8+2+4+4 = 61. So the batch header is 61 bytes, and that value is used all over the code under the name RECORD_BATCH_OVERHEAD. Here is how it maps to what you see in your app.

The two blue boxes are the fields the broker fills in; the yellow area is covered by the CRC. Both broker-filled fields sit before the CRC.

The two fields the CRC doesn't cover

The thing to notice in the picture is where the CRC sits. The comment even gives the reason.

 * The CRC covers the data from the attributes to the end of the batch (i.e. all the bytes that follow the CRC). It is
 * located after the magic byte, which means that clients must parse the magic byte before deciding how to interpret
 * the bytes between the batch length and the magic byte. The partition leader epoch field is not included in the CRC
 * computation to avoid the need to recompute the CRC when this field is assigned for every batch that is received by
 * the broker. The CRC-32C (Castagnoli) polynomial is used for the computation.

clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java lines 67–71

The CRC is computed only from attributes to the end of the batch. baseOffset and partitionLeaderEpoch, which come before it, are fields the broker fills in again for every batch it receives; keeping them outside the CRC means the broker can change them without recomputing the CRC. The computeChecksum() method looks at exactly that range in one line, Crc32C.compute(buffer, ATTRIBUTES_OFFSET, buffer.limit() - ATTRIBUTES_OFFSET) (lines 409–411).

There is also a reason magic comes before the CRC. In the old formats (v0, v1) the bytes in the same positions mean different things, so a reader has to look at magic first to decide how to read the rest (comment lines 67–69). This "fields the broker changes live outside the CRC" layout is what makes Section 5's "write it almost as-is" possible.

Example 1: looking at a batch on disk with kafka-dump-log.sh

With an ordinary KafkaProducer I sent three records and called flush(), then sent two more and called flush() again. The first record carries the header trace=abc. Then I decoded the partition's segment file (00000000000000000000.log) with kafka-dump-log.sh --deep-iteration --print-data-log.

These examples were run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411). The example folder is ex/u06-produce-path, run on a single node with the broker and controller in one process (kafka_single). The core of the sending code:

// Written for this post (not part of Kafka's source)
// Sends records with an ordinary KafkaProducer. run.sh looks at what the broker wrote to disk with kafka-dump-log.sh.
// ...
                    for (int i = 0; i < 3; i++) {
                        ProducerRecord<String, String> r = new ProducerRecord<>(topic, 0, "k" + i, "hello-" + i);
                        if (i == 0) r.headers().add("trace", "abc".getBytes(StandardCharsets.UTF_8));
                        fs.add(p.send(r));
                    }
                    p.flush();   // send the first batch
                    for (int i = 3; i < 5; i++) fs.add(p.send(new ProducerRecord<>(topic, 0, "k" + i, "hello-" + i)));
                    p.flush();   // the second batch

=== 1. a normal producer: 3 records, flush, 2 records, flush
acked: partition 0 offset 0
acked: partition 0 offset 1
acked: partition 0 offset 2
acked: partition 0 offset 3
acked: partition 0 offset 4
--- kafka-dump-log.sh --deep-iteration --print-data-log
baseOffset: 0 lastOffset: 2 count: 3 baseSequence: 0 lastSequence: 2 producerId: 0 producerEpoch: 0 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 0 CreateTime: 1794025864682 size: 119 magic: 2 compresscodec: none crc: 199030714 isvalid: true
 offset: 0 CreateTime: 1794025864656 keySize: 2 valueSize: 7 sequence: 0 headerKeys: [trace] key: k0 payload: hello-0
 offset: 1 CreateTime: 1794025864682 keySize: 2 valueSize: 7 sequence: 1 headerKeys: [] key: k1 payload: hello-1
 offset: 2 CreateTime: 1794025864682 keySize: 2 valueSize: 7 sequence: 2 headerKeys: [] key: k2 payload: hello-2
baseOffset: 3 lastOffset: 4 count: 2 baseSequence: 3 lastSequence: 4 producerId: 0 producerEpoch: 0 partitionLeaderEpoch: 0 isTransactional: false isControl: false deleteHorizonMs: OptionalLong.empty position: 119 CreateTime: 1794025864917 size: 93 magic: 2 compresscodec: none crc: 3096572621 isvalid: true
 offset: 3 CreateTime: 1794025864917 keySize: 2 valueSize: 7 sequence: 3 headerKeys: [] key: k3 payload: hello-3
 offset: 4 CreateTime: 1794025864917 keySize: 2 valueSize: 7 sequence: 4 headerKeys: [] key: k4 payload: hello-4

Timestamps and CRCs differ on every run.

1. There are two batches. Just as the flush() calls split them, there is one batch of three records (offsets 0–2) and one of two (3–4). The second batch's position: 119 means the first batch occupies bytes 0–118 of the file.
2. partitionLeaderEpoch: 0 — the producer doesn't fill this field (as Example 3 shows, it sends -1). The broker wrote leader epoch 0.
3. producerId: 0, baseSequence: 0 / 3 — without any configuration, the idempotent producer's ticket is attached. That's because enable.idempotence defaults to true at this commit (the define starting at ProducerConfig.java line 574). The second batch's baseSequence is 3 because, like offsets, the sequence grows by one per record.
4. The CreateTime on a batch line is the header's maxTimestamp; the CreateTime on the record lines starting with | is each record's own timestamp.

Example 2: reading the first batch byte by byte

kafka-dump-log.sh decodes batches with Kafka's own classes. To see whether the bytes really sit in the order of the table above, I wrote a program (BatchHeader.java) that reads the file directly, without Kafka's classes, and prints each field's position and bytes. The field order and sizes copy the constants above.

=== 2. the first batch, byte by byte
--- batch header (61 bytes)
 0- 7 baseOffset            0               00 00 00 00 00 00 00 00
 8-11 batchLength           107             00 00 00 6b
12-15 partitionLeaderEpoch  0               00 00 00 00
16-16 magic                 2               02
17-20 crc                   199030714       0b dc f7 ba
21-22 attributes            0               00 00
23-26 lastOffsetDelta       2               00 00 00 02
27-34 baseTimestamp         1794025864656   00 00 01 a1 b4 46 3d d0
35-42 maxTimestamp          1794025864682   00 00 01 a1 b4 46 3d ea
43-50 producerId            0               00 00 00 00 00 00 00 00
51-52 producerEpoch         0               00 00
53-56 baseSequence          0               00 00 00 00
57-60 recordsCount          3               00 00 00 03
--- first record (varints)
61-61 length                25              32
62-62 attributes            0               00
63-63 timestampDelta        0               00
64-64 offsetDelta           0               00
65-65 keyLength             2               04
66-67 key                   "k0"            6b 30
68-68 valueLength           7               0e
69-75 value                 "hello-0"       68 65 6c 6c 6f 2d 30
76-76 headersCount          1               02
77-77 headerKeyLength       5               0a
78-82 headerKey             "trace"         74 72 61 63 65
83-83 headerValueLength     3               06
84-86 headerValue           "abc"           61 62 63
next record starts at byte 87

This output comes from the same run as Example 1; the crc and timestamp values and bytes differ on every run. Looking only at the header (the record part is for Section 2):

• Bytes 0–60 are the 61-byte header, and the first record starts at byte 61. The crc value matches crc: 199030714 from Example 1.
• All 8 bytes of baseOffset are 0. This is the first batch in the file, so 0 is right. For the second batch this field would be 3 (baseOffset: 3 in Example 1).

Easy to get wrong: batchLength is not the batch size

Example 1's size: 119 and Example 2's batchLength 107 differ. batchLength is the number of bytes after itself. It leaves out the 12 bytes of baseOffset (8) and batchLength (4) in front, so 119 - 12 = 107. These 12 bytes are the "offset + size" prefix that every format since v0 keeps in the same place, and they have a name: Records.LOG_OVERHEAD (Records.java line 49). Code that scans a file can jump to the next batch by reading just these 12 bytes, without knowing the format.

2. Records — saving bytes with varints

The fields of a record

A single record inside a batch is handled by DefaultRecord. Its format is again in the class comment.

 * Record =>
 *   Length => Varint
 *   Attributes => Int8
 *   TimestampDelta => Varlong
 *   OffsetDelta => Varint
 *   KeyLength => Varint
 *   Key => Bytes
 *   ValueLength => Varint
 *   Value => Bytes
 *   HeadersCount => Varint
 *   Headers => [HeaderKey HeaderValue]
 *     HeaderKeyLength => Varint
 *     HeaderKey => String
 *     HeaderValueLength => Varint
 *     HeaderValue => Bytes

clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecord.java lines 40–54

Unlike the batch header, almost every field is a Varint/Varlong. And the offset and timestamp are not written as-is but as the difference from the batch's baseOffset/baseTimestamp (OffsetDelta, TimestampDelta). Offset differences inside a batch are 0, 1, 2 …, and timestamp differences are usually a few milliseconds, which suits varints, where small numbers are short. The writing code, DefaultRecord.writeTo(), simply calls ByteUtils.writeVarint()/writeVarlong() in this order (lines 168–223). With no key (null) it writes -1 in the length slot (lines 183–184). That's why "empty key" (length 0) and "no key" (-1) can be told apart.

Zigzag varints

The comment on ByteUtils.writeVarint() says it follows the same scheme as Protocol Buffers, and its body is the single line writeUnsignedVarint((value << 1) ^ (value >> 31), out) (ByteUtils.java lines 474–476). Two steps:

1. Zigzag: (value << 1) ^ (value >> 31) turns 0, -1, 1, -2, 2 … into 0, 1, 2, 3, 4 …. The point is that negative numbers (like -1 meaning "none") also become small positive numbers.
2. Varint: the result is cut into 7-bit pieces, one per byte, with the top bit meaning "another byte follows". 0–63 become 0–126 after zigzag and fit in one byte.

The record part of Example 2 is exactly this.

The first record as read in Example 2. Left of the arrow are the bytes in the file, right of it the value after undoing zigzag.

• The length byte 32 (decimal 50) un-zigzags to 25, and indeed the 25 bytes 62–86 are the rest of this record.
• keyLength 2 is 04, valueLength 7 is 0e, headersCount 1 is 02 — each a single byte holding "value × 2".
• The first record is the batch's reference point, so its timestampDelta and offsetDelta are both 0.

The fixed cost of one record, not counting key, value and headers, is at most 21 bytes (MAX_RECORD_OVERHEAD, DefaultRecord.java lines 71–72). In this example it was 7 bytes once the 9 bytes of key and value are taken out (the 10 header bytes aside). In exchange, the "shared parts" (offset base, timestamp base, producer ticket, CRC) appear once in the batch header. The more records you gather, the more ways those 61 bytes are split, so making batches bigger with linger.ms and batch.size from Part 5 pays off on both disk and network.

Easy to get wrong: same records, different batch size on each run

The first time I ran Example 1 the first batch was 119 bytes, but on another run the same keys and values produced 121 bytes. In that run the timestamp difference between the first record and the second and third was 108 ms. 108 becomes 216 after zigzag, which no longer fits in one byte (max 127), so timestampDelta grew by one byte in two records. With varints, the value changes the size. The same data can take a few bytes more or less on disk depending on the time gaps between records.

3. The path of a Produce request inside the broker

The records arrive as a slice of the request buffer

In the JSON schema for the Produce request we saw in Part 3, the slot that carries the batch has a special type, records: { "name": "Records", "type": "records", … } (ProduceRequest.json lines 72–73). ProduceRequestData, generated from this schema at build time, reads that slot with _readable.readRecords(length) (line 906 of org/apache/kafka/common/message/ProduceRequestData.java, generated from the reference commit). The method lives in the Readable interface.

    default MemoryRecords readRecords(int length) {
        if (length < 0) {
            // no records
            return null;
        } else {
            ByteBuffer recordsBuffer = readByteBuffer(length);
            return MemoryRecords.readableRecords(recordsBuffer);
        }

clients/src/main/java/org/apache/kafka/common/protocol/Readable.java lines 65–72

readByteBuffer() returns a slice() of the received request buffer (ByteBufferAccessor.java lines 74–81). Instead of copying the bytes into a new array, it builds a MemoryRecords that points straight at that stretch of the buffer received from the network. This parsing happens when the Processor from Part 4 calls new Request(...) right after a full request has arrived. It pairs with what we saw in Part 3, where the producer attaches the records buffer to the request without copying it. The bytes the producer built end up sitting in the broker's memory unchanged.

The whole flow

Once a request handler thread enters case ApiKeys.PRODUCE => handleProduceRequest(request, requestLocal) in KafkaApis.handle() (KafkaApis.scala line 175), the path looks like this.

One request handler thread does everything from step 2 to step 14 in one go (it breaks off at step 12 when acks=all has to wait; for a transactional batch it can also break off inside step 3, between handleProduceAppend and appendRecords, while waiting for the coordinator's check — see "Easy to get wrong" at the end of this section).

Of steps 1–14, this section covers 2–6, Sections 4, 5 and 6 cover 7–10, and Section 6 covers 11–14.

handleProduceRequest() — checks that don't open the batch

KafkaApis.handleProduceRequest() checks authorization and topic existence for each topic-partition and gathers only those that pass.

    topicIdToPartitionData.foreach { case (topicIdPartition, partition) =>
      // ... (three KAFKA-10698 comment lines)
      val memoryRecords = partition.records.asInstanceOf[MemoryRecords]
      if (!authorizedTopics.contains(topicIdPartition.topic))
        unauthorizedTopicResponses += topicIdPartition -> new PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)
      else if (!metadataCache.contains(topicIdPartition.topicPartition))
        nonExistingTopicResponses += topicIdPartition -> new PartitionResponse(Errors.UNKNOWN_TOPIC_OR_PARTITION)
      else
        try {
          ProduceRequest.validateRecords(request.header.apiVersion, memoryRecords)
          authorizedRequestInfo += (topicIdPartition -> memoryRecords)
        } catch {
          case e: ApiException =>
            invalidRequestResponses += topicIdPartition -> new PartitionResponse(Errors.forException(e))
        }
    }

core/src/main/scala/kafka/server/KafkaApis.scala lines 438–455

Reading the Scala — xs.foreach { case (a, b) => … } is a lambda that takes a pair (tuple) and unpacks it into two names. val is a variable that is never reassigned (Java's final), and x.asInstanceOf[T] is the cast (T) x. m += (k -> v) puts into a mutable map, and try { … } catch { case e: ApiException => … } is the same as Java's catch (ApiException e).

ProduceRequest.validateRecords() is the first to look at the shape of the batch.

    public static void validateRecords(short version, BaseRecords baseRecords) {
        if (baseRecords instanceof Records) {
            Records records = (Records) baseRecords;
            Iterator<? extends RecordBatch> iterator = records.batches().iterator();
            if (!iterator.hasNext())
                throw new InvalidRecordException("Produce requests with version " + version + " must have at least " +
                        "one record batch per partition");

            RecordBatch entry = iterator.next();
            if (entry.magic() != RecordBatch.MAGIC_VALUE_V2)
                throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " +
                        "contain record batches with magic version 2");
            // ... (the check that rejects zstd below v7)
            if (iterator.hasNext())
                throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " +
                        "contain exactly one record batch per partition");
        }
    }

clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java lines 211–232

It checks three things: is there at least one batch, is magic 2, and is there exactly one batch per partition. The producer's Sender puts only one batch per partition in a request (Part 5), so normally this never trips. A partition that fails here gets INVALID_RECORD and is not written. At this stage only batch headers are read; the records are not unpacked.

Next comes replicaManager.handleProduceAppend(...) (lines 548–557). If there are transactional batches, this method first asks the transaction coordinator "has this partition been added to the transaction?" (Part 12); otherwise it goes straight to appendRecords(). The first check in appendRecords() is the acks value. isValidRequiredAcks() is the one-liner requiredAcks == -1 || requiredAcks == 1 || requiredAcks == 0 (lines 1369–1371); for anything other than -1, 0 or 1 nothing is written and every partition gets INVALID_REQUIRED_ACKS (lines 648–651). The producer's acks setting only accepts all, -1, 0 and 1 (ProducerConfig.java lines 430–435), so an ordinary producer can never see this error; we confirm it in Example 3 with a hand-built request. If the value passes, appendToLocalLog() finds the Partition object for each partition and calls partition.appendRecordsToLeader(records, origin, requiredAcks, …) (lines 1414–1416).

Partition.appendRecordsToLeader() — am I the leader, is the ISR big enough?

  def appendRecordsToLeader(
    records: MemoryRecords,
    // ... (origin, requiredAcks, requestLocal and so on)
  ): LogAppendInfo = {
    val (info, leaderHWIncremented) = inReadLock(leaderIsrUpdateLock, () => {
      leaderLogIfLocal match {
        case Some(leaderLog) =>
          val minIsr = effectiveMinIsr(leaderLog)
          val inSyncSize = partitionState.isr.size

          // Avoid writing to leader if there are not enough insync replicas to make it safe
          if (inSyncSize < minIsr && requiredAcks == -1) {
            throw new NotEnoughReplicasException(s"The size of the current ISR : $inSyncSize " +
              s"is insufficient to satisfy the min.isr requirement of $minIsr for partition $topicPartition, " +
              s"live replica(s) broker.id are : $inSyncReplicaIds")
          }

          val info = leaderLog.appendAsLeader(records, this.leaderEpoch, origin, requestLocal, verificationGuard, transactionVersion)

          // we may need to increment high watermark since ISR could be down to 1
          (info, maybeIncrementLeaderHW(leaderLog))

        case None =>
          throw new NotLeaderOrFollowerException("Leader not local for partition %s on broker %d"
            .format(topicPartition, localBrokerId))
      }
    })

    info.copy(if (leaderHWIncremented) LeaderHwChange.INCREASED else LeaderHwChange.SAME)
  }

core/src/main/scala/kafka/cluster/Partition.scala lines 1222–1255

Reading the Scala — val (info, leaderHWIncremented) = … receives a returned pair into two names. leaderLogIfLocal match { case Some(leaderLog) => … case None => … } branches on "if this broker has the leader log, take it out and use it; otherwise…"; Option/Some/None are like Java's Optional. inReadLock(lock, () => { … }) runs the lambda while holding the read lock and returns its value.

It does three things here.

1. Is this broker the leader? Without a leader log it's NotLeaderOrFollowerException. When the producer's Sender gets this error (a subclass of InvalidMetadataException) it asks for fresh metadata (from Sender.java line 713; for request v10 and later KafkaApis also puts the current leader in the response — lines 476–488).
2. With acks=-1, is the ISR big enough? If the ISR is smaller than min.insync.replicas, it throws NotEnoughReplicasException before writing. It doesn't write and then fail while waiting for replication; it refuses from the start.
3. It writes to the leader log with appendAsLeader() (Section 4) and raises the high watermark if it can (Section 6).

It is also worth noticing that the lock is a read lock. leaderIsrUpdateLock is taken as a write lock when the leader or the ISR changes, while write requests enter together under the read lock. The order of concurrent writes to the same partition is decided by UnifiedLog's lock further down (Section 4).

Example 7: following the write path in the broker log

To see that requests really take this path, and on which thread, I changed the broker's logging so that ReplicaManager, UnifiedLog and LogSegment at TRACE and the request logger (kafka.request.logger, Part 4) at DEBUG all go to one file with the thread name (%t). Instead of config/log4j2.yaml, the example folder's log4j2-broker.yaml is passed through KAFKA_LOG4J_OPTS (bin/kafka-server-start.sh only sets the default when that variable is empty). Below are only the lines for the first batch of Example 1, shortened (the example output numbers this section 7, but it belongs here in the story).

=== 7. the append path of the first batch in section 1, from the broker log (shortened)
data-plane-kafka-request-handler-7 | ReplicaManager | Append [u06-0 -> MemoryRecords(size=119, ...)] to local log
data-plane-kafka-request-handler-7 | LogSegment | Inserting 119 bytes at end offset 2 at position 0
data-plane-kafka-request-handler-7 | LogSegment | Appended 119 to /tmp/kafka/1/data/u06-0/00000000000000000000.log at end offset 2
data-plane-kafka-request-handler-7 | UnifiedLog | Appended message set with last offset: 2, first offset: 0, next offset: 3
data-plane-kafka-request-handler-7 | UnifiedLog | Setting high watermark (offset=3, segment=[0:119])
data-plane-kafka-request-handler-7 | ReplicaManager | 119 written to log u06-0 beginning at offset 0 and ending at offset 2
data-plane-kafka-request-handler-7 | ReplicaManager | Produce to local log in 34 ms
data-plane-kafka-network-thread-1-PLAINTEXT-PLAINTEXT-2 | logger | Completed request (PRODUCE v13, correlationId=6, acks=-1)

Thread numbers and timings differ on every run.

1. From Append to Produce to local log, every line comes from one thread, data-plane-kafka-request-handler-N. Steps 2–13 of the picture happen on a single request handler thread. On another run the first and second batches were handled by handler threads with different numbers — any thread may take the next request off the queue (Part 4).
2. The order is ReplicaManager (Append) → LogSegment (Inserting, Appended) → UnifiedLog (Appended message set) → UnifiedLog (Setting high watermark) → ReplicaManager (written to log). UnifiedLog's "Appended message set" is logged after writing to the segment, so it comes after LogSegment.
3. "Setting high watermark (offset=3 …)" — the topic has only one replica, so the high watermark moves up to the log end (3) right after the write (Section 6).
4. The final Completed request line is logged by a network thread (data-plane-kafka-network-thread-…) after it has finished sending the response. You can also see the producer's request is Produce v13 with acks=-1. The producer's acks default at this commit is all, which travels as -1 in the request.

Easy to get wrong

• KafkaApis doesn't call appendRecords() directly. It's tempting to say "handleProduceRequest() then replicaManager.appendRecords()", but at this commit (and in 4.4.0) handleProduceAppend() sits in between and does the transaction check first. When that check completes asynchronously, appendRecords() may be called on a different request handler thread (KafkaRequestHandler.wrapAsyncCallback, ReplicaManager.scala lines 791–794).
• The min.insync.replicas check only happens for acks=-1. An acks=1 write is accepted even if the ISR is just the leader.

4. UnifiedLog.append() — where offsets are assigned

The leader and the followers use the same append()

appendAsLeader() is a thin entrance.

    public LogAppendInfo appendAsLeader(MemoryRecords records,
                                        int leaderEpoch,
                                        AppendOrigin origin,
                                        RequestLocal requestLocal,
                                        VerificationGuard verificationGuard,
                                        short transactionVersion) {
        boolean validateAndAssignOffsets = origin != AppendOrigin.RAFT_LEADER;
        return append(records, origin, validateAndAssignOffsets, leaderEpoch, Optional.of(requestLocal),
            verificationGuard, false, RecordBatch.CURRENT_MAGIC_VALUE, transactionVersion);
    }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1062–1071

The real work is done by the private method append(). appendAsFollower(), which writes batches a follower has copied from the leader, calls the same append() but passes false as the third argument, validateAndAssignOffsets (lines 1093–1103). Only the leader assigns offsets; followers write the offsets the leader assigned, unchanged. That's why the same record has the same offset on every replica. RAFT_LEADER is the case of the KRaft metadata log (Part 10), which has already decided its offsets.

Everything from the yellow synchronized (lock) box down (to the end of both branches) is the stretch that only one append per partition log passes through at a time. Grey boxes belong to other posts.

Outside the lock: checks that don't open the batch

append() first scans the batches without a lock in analyzeAndValidateRecords(). These checks read only batch headers.

            if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 && origin == AppendOrigin.CLIENT && batch.baseOffset() != 0) {
                throw new InvalidRecordException("The baseOffset of the record batch in the append to " + topicPartition() + " should " +
                        "be 0, but it is " + batch.baseOffset());
            }

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1525–1528

The baseOffset of a v2 batch sent by a client must be 0. The producer doesn't know offsets, so it puts 0, and anything else is treated as a mistake. Next it checks whether the batch size exceeds the topic's max.message.bytes (lines 1564–1570) and whether the CRC matches (batch.isValid(), line 1573). The CRC check is one pass over the whole batch with the computeChecksum() we saw earlier.

Inside the lock: counting from the current log end

Once the checks pass, it takes the partition log's lock and assigns offsets.

            synchronized (lock)  {
                // ...
                            if (validateAndAssignOffsets) {
                                // assign offsets to the message set
                                PrimitiveRef.LongRef offset = PrimitiveRef.ofLong(localLog.logEndOffset());
                                appendInfo.setFirstOffset(offset.value);
                                Compression targetCompression = BrokerCompressionType.targetCompression(config().compression, appendInfo.sourceCompression());
                                LogValidator validator = new LogValidator(validRecords,
                                        // ... (topic settings)
                                        leaderEpoch,
                                        // ...
                                );
                                LogValidator.ValidationResult validateAndOffsetAssignResult = validator.validateMessagesAndAssignOffsets(offset,
                                        // ...
                                validRecords = validateAndOffsetAssignResult.validatedRecords();
                                appendInfo.setMaxTimestamp(validateAndOffsetAssignResult.maxTimestampMs());
                                appendInfo.setLastOffset(offset.value - 1);

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java lines 1150–1182

1. offset starts at the current log end offset (localLog.logEndOffset()). PrimitiveRef.LongRef is a box holding one long; LogValidator increments it for every record it counts.
2. The target compression (targetCompression) is decided from the topic's compression.type and the batch's compression (Section 5), and a LogValidator is created to do the checking and offset assignment. The leaderEpoch passed in becomes the batch header's partitionLeaderEpoch.
3. When it returns, offset.value - 1 is the last offset of this batch.

Reading the offset (line 1158), assigning offsets, writing the file and raising the log end (localLog.append(), line 1276) all happen inside the same synchronized (lock). Even if two request handler threads write to the same partition at once, one waits until the other finishes, so two batches can never get the same offset. The flip side is that writes to one partition line up single-file inside the broker. That is one reason partitions are the unit of write parallelism.

After that come the leader epoch cache entry (assignEpochStartOffset, Part 9), switching to a new segment if the current one is full (maybeRoll(), Part 7), and the duplicate/order check for idempotent and transactional producers (analyzeAndValidateProducerState(), Part 12). If that last check decides the batch was already written and has been sent again (a retry with the same sequence), nothing is written and the earlier offsets are returned as they are (lines 1261–1268). This is the spot where an idempotent producer's retries don't write records twice.

Example 3: requests the broker refuses

Sending a hand-built batch lets us reach branches an ordinary producer never takes. RawProduce.java builds a batch with MemoryRecords.withRecords(...) without a KafkaProducer, puts it into a generated ProduceRequestData, and sends a Produce v12 request (the last version that picks topics by name) over a plain socket. The request framing (4-byte length + header + body) is the same as in the Part 3 example. The core of it:

// Written for this post (not part of Kafka's source)
// Builds a record batch by hand, without KafkaProducer, puts it in a Produce request (v12) and sends it over a plain socket.
// Printing the batch header (baseOffset, partitionLeaderEpoch, crc) before sending lets us compare
// it with the batch the broker wrote to disk (kafka-dump-log.sh): what stayed the same and what changed.
// ...
            // the client fills baseOffset with a non-zero value (5)
            case "base-offset" -> records = MemoryRecords.withRecords(5L, Compression.NONE, recs("x"));
            case "two-batches" -> {
                // glue two batches together in one partition's slot
                ByteBuffer b1 = MemoryRecords.withRecords(Compression.NONE, recs("x")).buffer();
                ByteBuffer b2 = MemoryRecords.withRecords(Compression.NONE, recs("y")).buffer();
                ByteBuffer both = ByteBuffer.allocate(b1.remaining() + b2.remaining());
                both.put(b1).put(b2).flip();
                records = MemoryRecords.readableRecords(both);
            }
        // ... (print the batch header before sending)
        ProduceRequestData.TopicProduceDataCollection topics = new ProduceRequestData.TopicProduceDataCollection();
        topics.add(new ProduceRequestData.TopicProduceData()
            .setName(topic)
            .setPartitionData(List.of(new ProduceRequestData.PartitionProduceData().setIndex(0).setRecords(records))));
        ProduceRequestData req = new ProduceRequestData().setAcks(acks).setTimeoutMs(30000).setTopicData(topics);

I sent requests with acks set to 2, with baseOffset set to 5, with two batches glued into one partition's slot, and with acks=0.

=== 5. requests the broker refuses, and acks=0
--- acks2
client batch: baseOffset=0 lastOffset=0 partitionLeaderEpoch=-1 crc=382737714 compression=none size=80
response: acks=2 error=INVALID_REQUIRED_ACKS baseOffset=-1 errorMessage=null
--- base-offset
client batch: baseOffset=5 lastOffset=5 partitionLeaderEpoch=-1 crc=1050192955 compression=none size=80
response: acks=1 error=INVALID_RECORD baseOffset=-1 errorMessage=null
--- two-batches
client batch: baseOffset=0 lastOffset=0 partitionLeaderEpoch=-1 crc=2714997260 compression=none size=80
client batch: baseOffset=0 lastOffset=0 partitionLeaderEpoch=-1 crc=1139444227 compression=none size=80
response: acks=1 error=INVALID_RECORD baseOffset=-1 errorMessage=null
--- acks0
client batch: baseOffset=0 lastOffset=0 partitionLeaderEpoch=-1 crc=4188389087 compression=none size=80
acks=0: no response within 3s
--- broker log: why base-offset was refused
InvalidRecordException: The baseOffset of the record batch in the append to u06-bad-0 should be 0, but it is 5
--- u06-bad on disk (only the acks=0 batch)
baseOffset: 0 lastOffset: 0 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 ... position: 0 CreateTime: 1794025933720 size: 80 magic: 2 compresscodec: none crc: 4188389087 isvalid: true

CRCs and timestamps differ on every run.

1. acks=2 → INVALID_REQUIRED_ACKS. Caught by isValidRequiredAcks().
2. baseOffset=5 → INVALID_RECORD. The response's errorMessage is empty, but the reason is in the broker log — the very sentence from lines 1526–1527 above. This exception came from UnifiedLog, so it had already passed the KafkaApis and Partition checks.
3. Two batches → INVALID_RECORD. Caught by validateRecords() in KafkaApis. This path leaves no log line and just returns the error code.
4. acks=0 → no response at all within 3 seconds. Yet on disk that batch is written at offset 0. None of the three earlier requests was written (more on this in Section 6).

Easy to get wrong: the log end offset rises after the file write

The moment offsets are assigned (line 1158) and the moment the log end offset rises (inside localLog.append(), LocalLog.java lines 577–580) are different. If a check right after assignment fails (a duplicate batch, exceeding the segment size), the log end stays where it was and the next batch gets the same offsets again. So the offsets don't get "holes" this way (empty offsets caused by transaction markers or compaction are another story, for Parts 7 and 12).

5. Writing it almost as-is — LogValidator's in-place fixes

Two branches

The entrance of LogValidator.validateMessagesAndAssignOffsets() splits on compression. If "both the batch's compression (sourceCompressionType) and the target compression are NONE" it takes the uncompressed path, otherwise the compressed path (lines 144–157). Produce requests v3 and later only accept v2 batches (Section 3), so hasMatchingMagic(toMagic) is always true and an uncompressed batch goes to assignOffsetsNonCompressed(). That method walks the records one by one, checking them (a compacted topic without a key? a timestamp outside the allowed range?) and incrementing the offset counter, and then fixes only a few fields of the batch header.

            batch.setLastOffset(offsetCounter.value - 1);

            if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2)
                batch.setPartitionLeaderEpoch(partitionLeaderEpoch);

            if (batch.magic() > RecordBatch.MAGIC_VALUE_V0) {
                if (timestampType == TimestampType.LOG_APPEND_TIME)
                    batch.setMaxTimestamp(TimestampType.LOG_APPEND_TIME, now);
                else
                    batch.setMaxTimestamp(timestampType, maxBatchTimestamp);
            }

storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java lines 264–274

Here is what setLastOffset() and setPartitionLeaderEpoch() do.

    @Override
    public void setLastOffset(long offset) {
        buffer.putLong(BASE_OFFSET_OFFSET, offset - lastOffsetDelta());
    }

    @Override
    public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) {
        long currentMaxTimestamp = maxTimestamp();
        // We don't need to recompute crc if the timestamp is not updated.
        if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp)
            return;

        byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch(), hasDeleteHorizonMs());
        buffer.putShort(ATTRIBUTES_OFFSET, attributes);
        buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp);
        long crc = computeChecksum();
        ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc);
    }

    @Override
    public void setPartitionLeaderEpoch(int epoch) {
        buffer.putInt(PARTITION_LEADER_EPOCH_OFFSET, epoch);
    }

clients/src/main/java/org/apache/kafka/common/record/internal/DefaultRecordBatch.java lines 376–398

• setLastOffset(offset) writes offset - lastOffsetDelta into a single field, baseOffset. The offsetDelta inside each record is a difference from the base, so it needs no change. Section 2's "write differences" pays off again here — with absolute offsets, every record would have to be fixed.
• setPartitionLeaderEpoch() also overwrites a single field.
• Both fields come before the CRC, so the CRC is not recomputed (Section 1).
• setMaxTimestamp() touches a field inside the CRC, so it does recompute the CRC — but if the value is unchanged it does nothing. On a default CreateTime topic, the max timestamp the producer wrote and the value recounted from the records are the same, so it usually hits that return.

This buffer is the slice of the request buffer from Section 3. Without building a new buffer or rewriting the records, it fixes two or three fields of the bytes it received and passes them on to the next step (the file write). It returns the records it was given, unchanged, as the second argument of ValidationResult (lines 285–290).

Compressed batches: same codec, fix in place; different codec, rebuild

A compressed batch goes to validateMessagesAndAssignOffsetsCompressed(). To check the records it has to decompress them, but how the result is written is decided by a single flag, inPlaceAssignment.

        // No in place assignment situation 1
        boolean inPlaceAssignment = sourceCompressionType == targetCompression.type();
        // ...
        if (!inPlaceAssignment) {
            return buildRecordsAndAssignOffsets(offsetCounter, now, firstBatch, validatedRecords,
                uncompressedSizeInBytes);
        } else {
            // we can update the batch only and write the compressed payload as is;
            // again we assume only one record batch within the compressed set
            offsetCounter.value += validatedRecords.size();
            // there is only one batch in this path, so last offset can be viewed as shallowOffsetOfMaxTimestamp
            long lastOffset = offsetCounter.value - 1;
            firstBatch.setLastOffset(lastOffset);

            if (timestampType == TimestampType.LOG_APPEND_TIME)
                maxTimestamp = now;

            if (toMagic >= RecordBatch.MAGIC_VALUE_V1)
                firstBatch.setMaxTimestamp(timestampType, maxTimestamp);

            if (toMagic >= RecordBatch.MAGIC_VALUE_V2)
                firstBatch.setPartitionLeaderEpoch(partitionLeaderEpoch);

storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java lines 303–405

• If the batch's compression equals the target (and the format matches and the inner offsets are 0, 1, 2 …), the compressed record part is left alone and only header fields are fixed. Just as the comment says, it will "write the compressed payload as is". It still decompresses the records to look at them, but uses skipKeyValueIterator(), which skips keys, values and headers (lines 334–338).
• If not, buildRecordsAndAssignOffsets() rebuilds the batch from scratch in a new buffer with MemoryRecordsBuilder and recompresses it with the target codec (lines 417–455). This is the branch that costs the most CPU.

Green is the "write it almost as-is" path; red is the path that rebuilds the batch.

The target compression is decided by BrokerCompressionType.targetCompression(logCompression, producerCompressionType): the compression set on the topic if there is one, otherwise the compression the producer sent (logCompression.orElseGet(...), BrokerCompressionType.java lines 43–45). If the topic's compression.type is producer, LogConfig.getCompression() returns Optional.empty() (LogConfig.java lines 367–382), and then the producer's compression is the target. And producer is exactly the default of the topic's compression.type (ServerLogConfigs.java line 183). So if you leave the setting alone, whatever codec you send takes the in-place path. Conversely, if a topic pins a codec like compression.type=lz4 and the producer sends something else (or nothing), the broker decompresses and recompresses every batch (the one exception is a control batch — a transaction marker — written uncompressed, which is left as is; LogValidator.java lines 325–326).

Why "almost as-is"

Here is what this design buys.

• CPU: an uncompressed batch is only scanned, and a batch with the same codec is decompressed to check but not recompressed. Compression is done by the producers (usually many app servers) and the broker only checks.
• Memory copies: the slice of the request buffer is fixed up and handed straight to the file channel (Section 6), so the broker doesn't move record bytes from buffer to buffer.
• The same shape as the reader's: the bytes on disk are the bytes that will go out on the network. That is what makes zero-copy — sending from the file straight to the socket for consumers — possible, which we'll see in Part 8.

Example 4: is the hand-built batch's CRC the same on disk?

With RawProduce.java I sent an uncompressed batch (three records) to the same topic twice. It prints the client-side batch header before sending, and we compare it with the batch on disk afterwards.

=== 3. a hand-built batch (no compression), sent with Produce v12, acks=1
client batch: baseOffset=0 lastOffset=2 partitionLeaderEpoch=-1 crc=3388881269 compression=none size=120
response: acks=1 error=NONE baseOffset=0 errorMessage=null
client batch: baseOffset=0 lastOffset=2 partitionLeaderEpoch=-1 crc=839275597 compression=none size=120
response: acks=1 error=NONE baseOffset=3 errorMessage=null
--- on disk
baseOffset: 0 lastOffset: 2 count: 3 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 ... position: 0 CreateTime: 1794025884365 size: 120 magic: 2 compresscodec: none crc: 3388881269 isvalid: true
baseOffset: 3 lastOffset: 5 count: 3 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 ... position: 120 CreateTime: 1794025890717 size: 120 magic: 2 compresscodec: none crc: 839275597 isvalid: true

CRCs and timestamps differ on every run.

1. Second batch: the client's baseOffset=0 and partitionLeaderEpoch=-1 became baseOffset: 3 and partitionLeaderEpoch: 0 on disk. The response's baseOffset=3 agrees.
2. Yet the crc is exactly the value the client computed (839275597). The broker did not change a single byte in the CRC-covered range (byte 21 to the end). size is still 120 too.
3. The client-built batch's producerId: -1 marks a non-idempotent batch, and it stays on disk as it was.

Next I sent a gzip-compressed batch to two topics: one with the default compression.type=producer, the other created with compression.type=lz4.

=== 4. a gzip batch: topic compression.type=producer vs lz4
--- topic compression.type=producer
client batch: baseOffset=0 lastOffset=2 partitionLeaderEpoch=-1 crc=760403212 compression=gzip size=120
response: acks=1 error=NONE baseOffset=0 errorMessage=null
baseOffset: 0 lastOffset: 2 count: 3 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 ... position: 0 CreateTime: 1794025909378 size: 120 magic: 2 compresscodec: gzip crc: 760403212 isvalid: true
--- topic compression.type=lz4
client batch: baseOffset=0 lastOffset=2 partitionLeaderEpoch=-1 crc=3659826953 compression=gzip size=122
response: acks=1 error=NONE baseOffset=0 errorMessage=null
baseOffset: 0 lastOffset: 2 count: 3 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 ... position: 0 CreateTime: 1794025918729 size: 129 magic: 2 compresscodec: lz4 crc: 1373228294 isvalid: true

The CRCs, the timestamps and the sizes of the compressed batches differ on every run (the compressed record part contains timestamp deltas; in another run the sizes were 119 → 127).

1. The producer topic: the batch on disk is compresscodec: gzip, and both crc and size match the client side. The compressed payload was written as it came.
2. The lz4 topic: the batch on disk is compresscodec: lz4, and both the size (122 → 129) and the crc changed. The broker decompressed the gzip data and recompressed it with lz4. The client can't tell any difference — the response is NONE either way.

Example 5: the same 2000 records, changing only the compression

compression.type shows up directly in disk size. I sent 2000 similar-looking JSON string records to separate topics, changing only the producer's compression.type (the topics keep the default producer). batch.size was raised to 1 MB so they all fit in one batch.

=== 6. the same 2000 records with each compression.type (topic: producer)
sent 2000 records with compression.type=none
sent 2000 records with compression.type=gzip
sent 2000 records with compression.type=snappy
sent 2000 records with compression.type=lz4
sent 2000 records with compression.type=zstd
none      169536 bytes, batches: 1, codec on disk: none
gzip       16562 bytes, batches: 1, codec on disk: gzip
snappy     26213 bytes, batches: 1, codec on disk: snappy
lz4        27187 bytes, batches: 1, codec on disk: lz4
zstd       12015 bytes, batches: 1, codec on disk: zstd

The exact byte counts vary a little between runs. The strings in this example repeat nearly the same content, so they compress extremely well — ratios on real data depend on the data. In all five cases the broker wrote the producer's chosen codec to disk unchanged (codec on disk). Compressing on the producer shrinks the network traffic, the broker's disk, replication traffic and the traffic out to consumers, all to this size.

Easy to get wrong: "no compression" can get compressed too

The producer's compression.type defaults to none (ProducerConfig.java line 436). If a topic sets something like compression.type=zstd, even an uncompressed batch fails the targetCompression.type() == NONE test at the entrance and takes the compressed path, and since what was sent (none) differs from the target (zstd), the broker compresses it. "Putting compression on the topic will shrink the disk" is true, but remember the broker pays that CPU. Also, differences within the same codec, such as the gzip level, don't trigger recompression — inPlaceAssignment only compares the kind of codec (type()) (line 304).

6. Writing the file, and acks

LogSegment.append() — straight into the file channel

The MemoryRecords, now checked and given offsets, goes through localLog.append() to the active segment (the last segment file, the one currently being written).

    public void append(long largestOffset,
                       MemoryRecords records) throws IOException {
        if (records.sizeInBytes() > 0) {
            LOGGER.trace("Inserting {} bytes at end offset {} at position {}",
                records.sizeInBytes(), largestOffset, log.sizeInBytes());
            int physicalPosition = log.sizeInBytes();

            ensureOffsetInRange(largestOffset);

            // append the messages
            long appendedBytes = log.append(records);
            LOGGER.trace("Appended {} to {} at end offset {}", appendedBytes, log.file(), largestOffset);
            // ... (per batch, add an index entry once index.interval.bytes is exceeded — Part 7)

storage/src/main/java/org/apache/kafka/storage/internals/log/LogSegment.java lines 250–261

log is the FileRecords wrapping the segment's .log file, and its append() writes the buffer to the FileChannel with records.writeFullyTo(channel) (FileRecords.java lines 244–252).

MemoryRecords.writeFullyTo() just repeats channel.write(buffer) until the whole buffer is written (MemoryRecords.java lines 85–92). The slice of the request buffer goes into the file channel as it is. Inserting 119 bytes … at position 0 and Appended 119 to …/00000000000000000000.log in Example 7 are the TRACE lines of these two steps.

There is no fsync (FileChannel.force()) here. write() only gets the data into the operating system's page cache; when it actually reaches the disk is up to the OS. The only flush condition at the end of UnifiedLog.append() is the single line if (localLog.unflushedMessages() >= config().flushInterval) flush(false); (UnifiedLog.java line 1300). flushInterval is the topic setting flush.messages, and its default is Long.MAX_VALUE (ServerLogConfigs.java line 101). With the default configuration there's no flush per write. Rather than making sure the data is on one machine's disk, Kafka protects data by copying it to several brokers — that is acks=all, the topic of the next section. Other flush points (when moving to a new segment, periodic flushes) are covered in section 2 of Part 7.

Three places where acks changes the response

Red boxes reject before writing, green boxes finish right after writing, and the yellow box waits.

acks=-1 (all) — where it decides whether to wait. After writing to the local log, appendRecords() calls maybeAddDelayedProduce(), which checks this condition.

  // If all the following conditions are true, we need to put a delayed produce request and wait for replication to complete
  //
  // 1. required acks = -1
  // 2. there is data to append
  // 3. at least one partition append was successful (fewer errors than partitions)
  private def delayedProduceRequestRequired(requiredAcks: Short,
                                            entriesPerPartition: Map[TopicIdPartition, MemoryRecords],
                                            localProduceResults: Map[TopicIdPartition, LogAppendResult]): Boolean = {
    requiredAcks == -1 &&
    entriesPerPartition.nonEmpty &&
    localProduceResults.values.count(_.exception().isPresent) < entriesPerPartition.size
  }

core/src/main/scala/kafka/server/ReplicaManager.scala lines 1356–1367

Reading the Scala — private def f(...): Boolean = { … } is a method returning Boolean, and without any return the value of the last expression is the return value. In localProduceResults.values.count(_.exception().isPresent), the _ is a short lambda standing for "each element" (r -> r.exception().isPresent()).

If acks=-1, there is data to write, and at least one partition's write succeeded, it creates a DelayedProduce and puts it in the purgatory (lines 903–911). This object sends the response "once this partition's high watermark has caught up to the end of the batch just written". Partition.checkEnoughReplicasReachOffset(), which checks that condition, and how the high watermark rises are Part 9's topic, so here we only name them (server/src/main/java/org/apache/kafka/server/purgatory/DelayedProduce.java). With acks=1 the condition is false, so the else branch calls the response callback right away.

acks=0 — where no response is sent. When acks is 0, KafkaApis' response callback does something else instead of responding.

      if (produceRequest.acks == 0) {
        // no operation needed if producer request.required.acks = 0; however, if there is any error in handling
        // the request, since no response is expected by the producer, the server will close socket server so that
        // the producer client will know that some error has happened and will refresh its metadata
        if (errorInResponse) {
          // ... (one log line)
          requestChannel.closeConnection(request, new ProduceResponse(mergedResponseStatus.asJava).errorCounts)
        } else {
          // Note that although request throttling is exempt for acks == 0, the channel may be throttled due to
          // bandwidth quota violation.
          requestHelper.sendNoOpResponseExemptThrottle(request)
        }
      } else {
        requestChannel.sendResponse(request, new ProduceResponse(mergedResponseStatus.asJava, maxThrottleTimeMs, nodeEndpoints.values.toList.asJava))
      }

core/src/main/scala/kafka/server/KafkaApis.scala lines 512–533

• With no error it sends a NoOpResponse. Despite the name, it is a marker meaning "send nothing": the Processor writes nothing to the socket and starts reading that connection again (SocketServer.scala lines 921–930, the unmute from Part 4).
• With an error it closes the connection. For a producer that isn't waiting for a response, that's the only way to signal an error. The comment says so (lines 513–515).

The producer side matches. When building a Produce request, Sender passes acks != 0 in the expectResponse slot (Sender.java line 945). The "no response within 3s" in Example 3, and the batch that was on disk anyway, are the result of these two places. Even with acks=0 the broker checks and writes exactly the same way; it just doesn't report the result.

The effective min.insync.replicas

The minIsr in the ISR check from Section 3 is not the configured value as-is.

/**
 * When setting the min ISR, there is no restriction on it. Even if the value does not make sense to be larger than
 * the replication factor. In case there are such setting, the effective min ISR of min(replication factor, min ISR)
 * is returned here.
 */
  private def effectiveMinIsr(leaderLog: UnifiedLog): Int = {
      leaderLog.config.minInSyncReplicas.min(remoteReplicasMap.size + 1)
  }

core/src/main/scala/kafka/cluster/Partition.scala lines 241–248

Reading the Scala — a.min(b) is Math.min(a, b). In Scala you can call methods even on primitive types like Int.

It takes the smaller of the setting and "the number of replicas (the other replicas + 1)". As the comment says, min.insync.replicas may be set higher than the replication factor without complaint; it is clipped here instead.

Example 6: min.insync.replicas=2 on a single-replica topic

I created a topic with replication factor 1 and min.insync.replicas=2, and sent with acks=all.

=== 8. min.insync.replicas=2 on a topic with one replica, acks=all
acks=all ok: offset 0

It succeeded without NotEnoughReplicas. The effective value is min(2, 1) = 1 and the ISR size is 1 as well. It's easy to think "the config asks for 2, so it must fail". Watching NotEnoughReplicas actually happen when a broker goes down and the ISR shrinks is for Part 9, with a three-replica cluster.

Easy to get wrong

• Even with acks=all, a single replica looks as if it doesn't wait. The request in Example 7 had acks=-1, so a DelayedProduce was created. But tryCompleteElseWatch(), which puts it in the purgatory, first checks the completion condition before watching (from DelayedOperationPurgatory.java line 122), and with only one replica the high watermark has already moved up ("Setting high watermark" in Example 7), so it completes immediately.
• "Success" with acks=1 means the data is in the leader's page cache. As shown above, it waits neither for fsync nor for follower replication. If the leader dies right after responding, the record may not exist on any follower. That gap is why acks=all and min.insync.replicas are used together, which continues in Part 9.

What this post leaves out

• The state checks for idempotent and transactional producers (analyzeAndValidateProducerState(), ProducerStateManager·ProducerAppendInfo): UnifiedLog.java lines 1257–1268, storage/src/main/java/org/apache/kafka/storage/internals/log/ProducerStateManager.java — Part 12
• Verifying transactional partitions inside the Produce request (maybeSendPartitionsToTransactionCoordinator in handleProduceAppend(), AddPartitionsToTxnManager): ReplicaManager.scala lines 697–833 — Part 12
• DelayedProduce and how the high watermark rises (Partition.checkEnoughReplicasReachOffset(), maybeIncrementLeaderHW()): server/src/main/java/org/apache/kafka/server/purgatory/DelayedProduce.java, from Partition.scala line 1013 — Part 9
• Follower writes (appendAsFollower()) and the leader epoch cache: UnifiedLog.java lines 1093–1103 — Part 9
• Segment rolling (maybeRoll()) and adding offset/time index entries: from UnifiedLog.java line 2166, the rest of LogSegment.append() — Part 7
• How the broker overwrites timestamps on a LogAppendTime topic (message.timestamp.type), and the allowed timestamp range (message.timestamp.before.max.ms/after.max.ms): LogValidator.java lines 570–598
• The second max.message.bytes check after recompression: UnifiedLog.java lines 1188–1201
• Quotas (throttling responses for write bandwidth and request quotas): KafkaApis.scala lines 495–509, server/src/main/java/org/apache/kafka/server/quota/ — not covered in this series
• The KRaft metadata log writing to the same UnifiedLog (AppendOrigin.RAFT_LEADER): raft/src/main/java/org/apache/kafka/raft/internals/KafkaRaftLog.java — Part 10
• Tiered storage, multiple log directories (JBOD), share groups — not covered in this series

Summary

What's next

We've followed a batch until it is appended to the end of the active segment's .log file. But there isn't just one file: the log is split into several segments, each accompanied by .index and .timeindex files, and old segments are deleted according to the retention settings or shrunk by compaction. maybeRoll(), which we only named in Section 4, and the index entries we skipped in LogSegment.append() are where that begins. It continues in Part 7: Log Storage — Segments, Indexes, Retention and Compaction.

What this post deferred "to Part 9" — what acks=all waits for in the purgatory, and how followers' Fetch requests move the high watermark up — is in Part 9: Replication — The ISR, the High Watermark and the Wait Behind acks=all.

• 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 5: Inside the Producer — From send() to Batches and the Sender Thread
• Part 8: The Read Path — Fetch Requests, Long Polling and Zero-Copy