Producers: Batching and the Record Batch

After this lesson your producer collects records per partition, decides when a batch is full or has waited long enough, and encodes each batch in the exact byte format a broker stores on disk and sends to consumers: Kafka's RecordBatch, magic 2.

Why producers batch

A request to a broker has a fixed cost: a network round trip, a header, a disk append. Sending one record per request pays that cost per record. A producer therefore keeps one open batch per partition and sends it when either:

• the next record would not fit within batch.size (default 16384 bytes), or
• the batch has waited linger.ms since its first record.

The first record always goes in, even if it alone is bigger than batch.size; it then travels as a batch of one. The size that counts is the encoded batch size: the 61-byte header plus every encoded record.

The batch is the unit of everything

Brokers do not store records one by one. They store batches, byte for byte as the producer built them, and consumers receive the same bytes. That is why the CRC, the producer id and the sequence number live in the batch header.

The CRC deliberately starts after the fields a broker rewrites (baseOffset, partitionLeaderEpoch), so assigning offsets never forces a re-checksum. CRC-32C is the Castagnoli polynomial (reflected form 0x82F63B78, initial value and final xor all ones); the check value of the ASCII text 123456789 is 0xe3069283.

One record

length          varint   size of everything after it
attributes      int8     0
timestampDelta  varint   record timestamp - baseTimestamp
offsetDelta     varint   index of the record in the batch (0, 1, 2 ...)
keyLength       varint   -1 means a null key
key             bytes
valueLength     varint
value           bytes
headers count   varint   0 in this course

Integers inside records are zigzag varints: map the signed value to unsigned so small magnitudes stay small (0, -1, 1, -2, 2 become 0, 1, 2, 3, 4; in general (n << 1) XOR (n >> 63) with an arithmetic shift), then write 7 bits at a time, lowest group first, setting the top bit of every byte except the last.

 0 -> 00        -1 -> 01       2 -> 04      63 -> 7e
64 -> 80 01   300 -> d8 04

Worked trace, a record with key id, value ok, timestamp delta 0, index 0:

body:  00  00  00  04 69 64  04 6f 6b  00
       |   |   |   |         |         +- headers count 0
       |   |   |   |         +- valueLength zigzag(2) = 04, then "ok"
       |   |   |   +- keyLength zigzag(2) = 04, then "id"
       |   |   +- offsetDelta 0
       |   +- timestampDelta 0
       +- attributes
length = 10 bytes -> zigzag(10) = 20 = 0x14
record = 14 00 00 00 04 69 64 04 6f 6b 00

Idempotence in the header

When a producer is idempotent it has a producer id and epoch from the broker, and numbers its records per partition. The batch header carries the producer id, epoch and the sequence number of its first record; the next batch for the same partition starts at the previous base sequence plus its record count. Sequences are per partition, not global. A later lesson builds the broker side.

Common mistakes

• Counting characters, not bytes, for key and value lengths.
• Unsigned varints. The lengths and deltas are zigzag encoded: a length of 70 is 8c 01, not 46.
• CRC over the wrong range. It starts at attributes, not at the start of the batch, and it is computed last, over the finished bytes.
• Rolling a batch too late. Compare the size the batch WOULD have with the new record, and seal first if that is over the limit.

Your exercise

The starter reads the commands and calls send and flush for you. Implement the encoders and the batching rules. The most likely mistake is the size check: the first record of a batch is always accepted.