What is Batch in Kafka Producer

Rather than making an expensive OS socket write system call for every individual invocation of producer.send(), the Kafka client serializes records into an internal in-memory ac...

🟢 Junior Level

What is Batching in Kafka Producer?

Batching in the Apache Kafka Producer is the architectural mechanism of aggregating multiple individual records (ProducerRecord) destined for the same topic partition into a single in-memory data buffer (ProducerBatch) before transmitting them across the network via a single TCP socket round-trip (RTT).

Rather than making an expensive OS socket write system call for every individual invocation of producer.send(), the Kafka client serializes records into an internal in-memory accumulator. A dedicated background I/O thread (Sender) drains these accumulated batches and dispatches them to the partition Leader broker in consolidated network requests.

Without Batching (RPS = 10,000):
App Thread ──send()──> TCP Packet 1 (IP + TCP + Kafka Header ~100B + Payload 100B) ──> Broker
App Thread ──send()──> TCP Packet 2 ────────────────────────────────────────────────> Broker
App Thread ──send()──> TCP Packet 3 ────────────────────────────────────────────────> Broker
Result: 10,000 network interruptions, 10,000 disk fsync operations, massive packet overhead.

With Producer Batching:
App Thread 1 ──send()─┐
App Thread 2 ──send()─┼─> [RecordAccumulator: ProducerBatch 64KB] ──Sender Thread──> 1 Consolidated TCP Request ──> Broker
App Thread 3 ──send()─┘
Result: Optimal network saturation, sequential block writes to disk, multi-record dictionary compression.

Why Batching is Indispensable

Dramatic Throughput Surge: Network protocol headers (IP, TCP, Kafka protocol framing) are amortized across thousands of records simultaneously.
High-Ratio Block Compression: Streaming compression algorithms (LZ4, ZSTD, Snappy, GZIP) rely on shared dictionary patterns. Compressing an isolated 200-byte JSON payload yields negligible savings (and can even increase size due to framing headers). Compressing a 64 KB batch of 300 JSON records yields a 3x–7x compression ratio, slashing network egress and OS PageCache consumption.
Optimized Broker Disk I/O: Brokers append the incoming batch directly into the partition .log segment via sequential file channel writes, preventing physical file fragmentation on broker disks.

Basic Configuration Parameters

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092,kafka-broker2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 1. Batch byte size limit per partition (default: 16,384 bytes = 16 KB)
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536); // 64 KB recommended for high throughput

// 2. Artificial accumulation delay (default: 0 ms)
props.put(ProducerConfig.LINGER_MS_CONFIG, 10); // Wait up to 10 ms for batch to fill

// 3. Total memory pool available to accumulator (default: 33,554,432 bytes = 32 MB)
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864L); // 64 MB

// 4. Batch compression algorithm
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

🟡 Middle Level

Message Lifecycle and Batch Transmission Triggers

When the application invokes producer.send(record, callback):

Producer interceptors execute.
Key and Value serializers convert objects into raw byte arrays (byte[]).
The Partitioner determines the destination partition.
The serialized payload is appended to the partition’s queue in the RecordAccumulator.

A ProducerBatch transitions to the ready state and is picked up by the Sender thread upon meeting any one of the following 5 conditions:

Capacity Threshold Reached (batch.size): Appending a new record fills the batch up to or beyond batch.size (e.g., 64 KB). The batch is immediately sealed and queued for dispatch.
Time Threshold Expired (linger.ms): The first record was appended at time $T_0$. If the batch has not filled to batch.size by $T_0 + \text{linger.ms}$, it is dispatched anyway, bounding end-to-end latency.
Buffer Memory Pressure (buffer.memory): When the free memory pool is fully depleted, the accumulator forces immediate drainage of incomplete batches to free memory buffers.
Explicit Application Flush (producer.flush()): The calling thread blocks until all in-flight and pending batches are dispatched and acknowledged.
Client Shutdown (producer.close()): All pending buffers undergo graceful draining within the specified timeout.

Trade-Off Matrix: `batch.size` vs. `linger.ms`

Memory Exhaustion: `buffer.memory` and `max.block.ms`

The memory allocated to buffer.memory (default: 32 MB) is shared across all partition queues managed by that KafkaProducer instance.

If a broker slows down, a network partition occurs, or the application produces faster than the network can drain:

Free memory slots in the internal BufferPool become exhausted.
Subsequent producer.send() calls block synchronously.
The blocking duration is bounded by max.block.ms (default: 60,000 ms / 1 minute).
If memory remains unavailable when max.block.ms expires, send() throws: org.apache.kafka.common.errors.TimeoutException: Failed to allocate memory within the configured max blocking time.

🔴 Senior Level

Under the Hood: `RecordAccumulator` and `BufferPool` Architecture

Within org.apache.kafka.clients.producer.internals, batching is governed by two core components: RecordAccumulator and BufferPool:

                 KafkaProducer.send() (Application Threads)
                            │
                            ▼
                   RecordAccumulator
                            │
           ConcurrentMap<TopicPartition, Deque<ProducerBatch>>
           ┌─────────────────────────────────────────────────┐
           │ "orders-0": [Batch 1 (full)] -> [Batch 2 (open)] │
           │ "orders-1": [Batch 1 (open)]                    │
           │ "users-0":  [Batch 1 (open)]                    │
           └─────────────────────────────────────────────────┘
                            │
                   Request Memory Buffer
                            │
                            ▼
                       BufferPool
           ┌─────────────────────────────────────────────────┐
           │ totalMemory = 32 MB                             │
           │ poolableSize = batch.size (16 KB)               │
           │ freeSlots: ArrayDeque<ByteBuffer> [16KB, 16KB..]│
           │ nonPooledAvailableMemory: long                  │
           │ waiters: ArrayDeque<Condition>                  │
           └─────────────────────────────────────────────────┘

BufferPool Mechanics and Zero-GC Memory Recycling

Buffer Recycling: To eliminate severe Garbage Collection pressure (allocating and destroying hundreds of thousands of byte arrays per second), Kafka manages its own off-heap/heap BufferPool.
When a batch requires exactly batch.size bytes, the pool delivers an existing pre-allocated ByteBuffer from its internal freeSlots (ArrayDeque<ByteBuffer>).
Once the broker acknowledges the batch, the client calls deallocate(). The buffer is cleared via buffer.clear() and returned back to freeSlots. Zero heap garbage is generated!
The Hazard of Oversized Records: If a single serialized message exceeds batch.size (e.g., a 250 KB record when batch.size = 64 KB):
- BufferPool cannot satisfy the request from its pooled freeSlots.
- The pool dynamically allocates an unpooled buffer via ByteBuffer.allocate(recordSize) from nonPooledAvailableMemory.
- Once dispatched, this buffer cannot be returned to freeSlots and is thrown away to the JVM Garbage Collector.
- If an application continually streams oversized records, it bypasses the buffer pool entirely, causing intense Eden/Old-Gen GC thrashing and Stop-the-World pauses.

Threading Model: Multiplexing in the Sender Thread

Kafka strictly decouples producer application concurrency from network execution:

Application Threads: Serialize records and append them into the thread-safe Deque<ProducerBatch> for their designated TopicPartition. Contention is minimal and localized per partition deque.
Background Sender Thread (Daemon):
- Iterates over the RecordAccumulator to identify all ready batches.
- Groups ready batches by Leader Broker (NodeId). If Broker 1 hosts partition leaders orders-0, orders-1, and payments-2, their respective batches are multiplexed into a single consolidated network ClientRequest (ProduceRequest).
- Passes the request to the underlying non-blocking Java NIO Selector.

Eliminating Partition Fragmentation: The Uniform Sticky Partitioner

Prior to Kafka 2.4, messages published without a key (key == null) used round-robin distribution: record 1 went to partition 0, record 2 to partition 1, etc.

The Batching Bottleneck: In a topic with 50 partitions and batch.size = 16 KB, the producer had to generate $50 \times 16 \text{ KB} = 800 \text{ KB}$ before any batch could fill. In low-to-medium traffic, records trickled out as single-message packets triggered by linger.ms, devastating compression and throughput.

Introduced in Kafka 2.4 (KIP-480) and refined in Kafka 3.3 (KIP-794), the Uniform Sticky Partitioner solved this:

Unkeyed records “stick” to a single active partition until its batch fills to batch.size or linger.ms expires.
Once that batch is sealed and dispatched, the partitioner rotates to the next available partition.
Impact: Decreases p99 latency by 30%–50% and raises batch fill density to 95%+ without code changes.

Round-Robin (Legacy):
Record 1 ──> Partition 0 [rec1 (sparse)] ──linger.ms expires──> Dispatches 1 record
Record 2 ──> Partition 1 [rec2 (sparse)] ──linger.ms expires──> Dispatches 1 record
Record 3 ──> Partition 2 [rec3 (sparse)] ──linger.ms expires──> Dispatches 1 record

Sticky Partitioner (Modern Standard):
Record 1 ──┐
Record 2 ──┼─> Partition 0 [rec1, rec2, rec3, ... recN] ──batch.size filled──> Dispatches 64 KB
Record 3 ──┘
(Rotates to Partition 1 for the next full batch)

Idempotence and Sequence Numbers in Batches

When enable.idempotence = true (default since Kafka 3.0):

Every batch is assigned a unique ProducerId (PID), a ProducerEpoch, and a baseSequence number corresponding to the first record in the batch.
The broker tracks the highest 5 acknowledged sequences per partition (max.in.flight.requests.per.connection <= 5).
If a transient network drop occurs after the broker writes to disk but before the ACK reaches the client, the producer retries.
The broker recognizes (PID, Epoch, baseSequence) as already committed, replies with success, and discards duplicate storage, guaranteeing strictly-once partition persistence.

4 Tricky Questions

1. What happens if a single message exceeds `batch.size`, but is smaller than `max.request.size`?

Answer: The message is successfully sent, but incurs significant performance overhead:

BufferPool detects that the requested size exceeds poolableSize (batch.size).
It allocates a non-pooled, ad-hoc ByteBuffer using ByteBuffer.allocate(recordSize) from nonPooledAvailableMemory.
A dedicated ProducerBatch is instantiated around this single record, sealed immediately, and flagged as ready for the Sender thread.
After transmission and broker acknowledgment, the buffer cannot be recycled into freeSlots and is abandoned to the JVM Garbage Collector.
If the record exceeds max.request.size (default: 1 MB), the producer immediately throws RecordTooLargeException without attempting memory allocation.

2. An application configures `linger.ms = 100`, yet batches are dispatched every 0.5 ms without filling up to `batch.size`. What causes this?

Answer: Four common architectural causes explain this behavior:

Synchronous Invocation: The application executes producer.send(record).get(). Blocking on the Future prevents parallel accumulation; only one record ever exists in memory at a time.
Single In-Flight Slot Available (max.in.flight.requests.per.connection = 1): As soon as an in-flight request finishes, the Sender thread awakens on a socket selector event. If another record is already waiting in the accumulator, the sender will dispatch it immediately to keep the TCP pipeline occupied rather than sitting idle waiting for linger.ms.
Memory Pool Starvation (buffer.memory exhausted): When available memory hits zero, the accumulator forcibly awakens the Sender thread to drain partial batches immediately to unblock waiting application threads.
Explicit Flushing: Another application thread is periodically invoking producer.flush().

3. How do `delivery.timeout.ms`, `request.timeout.ms`, `linger.ms`, and retries interact?

Answer: Kafka 2.0 introduced delivery.timeout.ms (default: 120,000 ms / 2 minutes) as an overarching upper bound on total record lifespan. The configuration invariant is: $\text{delivery.timeout.ms} \ge \text{linger.ms} + \text{request.timeout.ms}$

Clock Start: The timer starts the instant the record enters the RecordAccumulator, not when it hits the network socket.
The record spends time waiting to batch (linger.ms or capacity), followed by time waiting in the socket queue.
If the broker rejects the batch with a retriable error (e.g., NOT_ENOUGH_REPLICAS), the batch returns to the head of the partition deque.
The producer continues retrying until the cumulative elapsed time exceeds delivery.timeout.ms. Once breached, the callback fails with a TimeoutException, regardless of how many retry attempts remain.

4. Why is compression (`compression.type`) applied per `ProducerBatch` rather than across the entire consolidated `ClientRequest`?

Answer: This design decision preserves Kafka’s end-to-end Zero-Copy broker architecture:

A single network ClientRequest can package batches destined for multiple disparate topics and partitions managed by the target broker.
If compression were applied to the entire request, the broker would be forced to decompress the entire payload into CPU memory, partition the records into separate log files, and recompress each partition log independently before writing to disk, exhausting broker CPU.
By compressing each ProducerBatch individually on the client:
- The broker performs lightweight header validation.
- The broker writes the compressed bytes directly into the partition log (.log) via OS PageCache without decompressing.
- When consumers read data, the broker streams the compressed segment directly from PageCache to the socket via the zero-copy sendfile(2) system call.
- Decompression is performed solely on the consumer client.

🎯 Interview Cheat Sheet

Core Concepts

Definition: Grouping records for the same partition into an in-memory buffer before sending via one TCP request.
Core Levers: batch.size (memory limit per partition, default: 16 KB) and linger.ms (wait window, default: 0 ms).
Trigger Rule: A batch dispatches as soon as batch.size is reached OR linger.ms expires.
Zero-Copy Integrity: Compression happens per batch so brokers can write compressed bytes directly to disk without CPU decompression.
Buffer Recycling: BufferPool recycles 16 KB/64 KB buffers via freeSlots to avoid generating JVM garbage.

Production Monitoring Metrics

Red Flags to Avoid

❌ “A batch aggregates messages across all topics and partitions into one buffer.” (Batches are partitioned strictly per TopicPartition).
❌ “Setting linger.ms = 0 means batching never occurs.” (Batches still form if network queues are occupied with in-flight requests).
❌ “The broker decompresses every batch to inspect messages before writing to disk.” (Brokers write compressed batches directly to disk segment files without decompression).
❌ “A 32 MB buffer means the producer can send a 32 MB message.” (Individual messages are bounded by max.request.size, defaulting to 1 MB).

How is Data Distributed Across Partitions — Sticky partitioner mechanics
What is Producer Acknowledgment and What Modes Exist (acks=0, 1, all) — Producer ACKs and Purgatory
How Does Message Compression Work — Compression algorithms and performance
What is Idempotent Producer — Sequence numbers and deduplication