KafkaIQ7.1 Producer Internals
Module Overview
Goal: Explain why configs behave the way they do by understanding internal mechanics.
What You'll Learn:
- Producer thread architecture
- RecordAccumulator and memory buffers
- Sender thread and network I/O
- Batching mechanics (how batches are formed)
- Metadata management and refresh
- Request lifecycle to brokers
- Memory management and back pressure
- Why certain configurations matter
What's NOT Covered Here:
- How to tune for performance (covered in Module 8)
- Business-level patterns (covered in Module 9.1)
- Error handling (covered in Module 6.1)
Key Insight: This module unlocks understanding, not usage.
1. Producer Architecture Overview
High-Level Components
┌─────────────────────────────────────────────────────────────┐
│ APPLICATION THREAD │
│ │
│ producer.send(record) ─────────────────────┐ │
│ ↓ │
│ ┌──────────────┐ │
│ │ Serializer │ │
│ └──────┬───────┘ │
│ ↓ │
│ ┌──────────────┐ │
│ │ Partitioner │ │
│ └──────┬───────┘ │
│ ↓ │
│ ┌──────────────────┐ │
│ │ RecordAccumulator│ │
│ │ (Memory Buffer) │ │
│ └────────┬─────────┘ │
└────────────────────────────────────────────┼──────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ SENDER THREAD (Background) │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Fetch Batches│ ──→ │ Send to │ │
│ │ from Buffer │ │ Broker │ │
│ └──────────────┘ └──────┬───────┘ │
│ │ │
│ ↓ │
│ ┌────────────────┐ │
│ │ Handle Response│ │
│ │ & Callbacks │ │
│ └────────────────┘ │
└─────────────────────────────────────────────────────────────┘
│
↓
┌─────────────────────────────────────────────────────────────┐
│ KAFKA BROKER │
└─────────────────────────────────────────────────────────────┘
2. Thread Model
Two Main Threads
1. Application Thread (User Thread)
Responsibilities:
- Serialize records
- Determine partition
- Add records to accumulator
- Return immediately (non-blocking)
What it does NOT do:
- Send data over network
- Wait for broker response
2. Sender Thread (Background I/O Thread)
Responsibilities:
- Fetch ready batches from accumulator
- Send batches to brokers
- Handle responses
- Invoke callbacks
- Manage retries
Lifecycle:
// Start
KafkaProducer producer = new KafkaProducer<>(props);
// → Sender thread starts automatically
// Shutdown
producer.close();
// → Sender thread stops after sending pending records
Thread Safety: KafkaProducer is Thread-Safe
Critical Point: KafkaProducer is fully thread-safe.
// ✅ SAFE: Multiple threads sharing one producer
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// Thread 1
producer.send(new ProducerRecord<>("topic", "key1", "value1"));
// Thread 2 (simultaneously)
producer.send(new ProducerRecord<>("topic", "key2", "value2"));
// Thread 3 (simultaneously)
producer.send(new ProducerRecord<>("topic", "key3", "value3"));
Why it's safe:
- The
send()method is synchronized internally - RecordAccumulator uses concurrent data structures
- Each thread's records are safely added to the buffer
- Sender thread coordinates access to batches
Best Practice: Share One Producer Across Threads
// ✅ GOOD: One producer, many threads
public class OrderService {
private final KafkaProducer<String, String> producer;
public OrderService() {
this.producer = new KafkaProducer<>(props);
}
// Called by multiple threads
public void publishOrder(Order order) {
producer.send(new ProducerRecord<>("orders", order.getId(), order.toJson()));
}
}
Why share one producer?
- Efficiency: One producer = one buffer, one sender thread, one set of connections
- Better batching: Records from all threads batch together
- Resource efficiency: Fewer TCP connections to brokers
- Simpler lifecycle: One producer to close
Anti-Pattern: One Producer Per Thread
// ❌ BAD: Creating producer per thread
public void publishOrder(Order order) {
KafkaProducer<String, String> producer = new KafkaProducer<>(props); // DON'T!
producer.send(new ProducerRecord<>("orders", order.getId(), order.toJson()));
producer.close(); // Expensive!
}
Problems:
- Each producer creates its own connections
- No batching across threads
- Overhead of creating/destroying producers
- Wastes memory and network resources
Exception: When Multiple Producers Make Sense
You might want separate producers when:
- Different configurations needed (e.g., different
ackslevels) - Isolation required (e.g., critical vs. non-critical data)
- Different clusters (e.g., multi-datacenter)
// ✅ OK: Different configs for different use cases
KafkaProducer<String, String> criticalProducer = new KafkaProducer<>(criticalProps);
KafkaProducer<String, String> analyticsProducer = new KafkaProducer<>(analyticsProps);
3. The RecordAccumulator
What is the RecordAccumulator?
The RecordAccumulator is an in-memory buffer that holds records before sending.
┌──────────────────────────────────────────────────────┐
│ RecordAccumulator (Memory Buffer) │
├──────────────────────────────────────────────────────┤
│ │
│ Topic: "orders" │
│ Partition 0: [batch1: record1, record2, ...] │
│ Partition 1: [batch2: record3, record4, ...] │
│ Partition 2: [batch3: record5, record6, ...] │
│ │
│ Topic: "events" │
│ Partition 0: [batch4: record7, record8, ...] │
│ Partition 1: [batch5: record9, record10, ...] │
│ │
└──────────────────────────────────────────────────────┘
Size limit: buffer.memory (default 32 MB)
Purpose
- Batch records for efficiency
- Decouple application thread from network I/O
- Buffer during temporary broker unavailability
- Enable async sends (application doesn't wait for network)
Buffer Structure
RecordAccumulator
├─ topic-A, partition-0 → Deque<ProducerBatch>
│ └─ [Batch 1] → [Batch 2] → [Batch 3] ...
│
├─ topic-A, partition-1 → Deque<ProducerBatch>
│ └─ [Batch 1] → [Batch 2] ...
│
└─ topic-B, partition-0 → Deque<ProducerBatch>
└─ [Batch 1] ...
Key: Each (topic, partition) has its own queue of batches.
4. Batching Mechanics
How Batches Are Created
Time Application Thread RecordAccumulator
─────┼────────────────────────────────┼──────────────────────────
0 │ send(record1) → partition 0 │
1 │ │ Create new batch
2 │ │ Add record1 to batch
3 │ │ Batch size: 100 bytes
4 │ │
5 │ send(record2) → partition 0 │
6 │ │ Add record2 to batch
7 │ │ Batch size: 200 bytes
8 │ │
9 │ send(record3) → partition 0 │
10 │ │ Add record3 to batch
11 │ │ Batch size: 300 bytes
│ │
│ ← Batch ready when: →
│ - Batch size ≥ batch.size (16 KB default)
│ - OR linger.ms expires (0ms default)
Batch Ready Conditions
A batch is ready to send when:
- Batch is full:
currentSize >= batch.size - Linger time expired: Time since first record ≥
linger.ms - Out of memory:
buffer.memoryexhausted (force send to free space) - Producer closing:
close()orflush()called
Example: Batching Timeline
Config:
batch.size = 16 KB
linger.ms = 10 ms
Timeline:
0ms │ record1 added (size: 1 KB) → Start linger timer
1ms │ record2 added (size: 2 KB) → Total: 3 KB
2ms │ record3 added (size: 3 KB) → Total: 6 KB
5ms │ record4 added (size: 4 KB) → Total: 10 KB
8ms │ record5 added (size: 5 KB) → Total: 15 KB
10ms │ Linger timer expires!
│ → Batch sent (15 KB < 16 KB, but linger.ms reached)
Result: Batch sent after 10ms with 5 records (15 KB).
Memory Management
Configuration
props.put("buffer.memory", 33554432); // 32 MB total buffer
props.put("batch.size", 16384); // 16 KB per batch
How Memory is Allocated
buffer.memory = 32 MB (total available)
Current usage:
topic-A, partition-0: 2 batches × 16 KB = 32 KB
topic-A, partition-1: 1 batch × 16 KB = 16 KB
topic-B, partition-0: 3 batches × 16 KB = 48 KB
...
Total used: 10 MB
Available: 22 MB
When Buffer is Full
// Application thread
producer.send(record); // Blocks!
// Blocks up to max.block.ms (default 60 seconds)
// If timeout → TimeoutException
Causes:
- Sending faster than broker can consume
- Network slow or broker down
- Batches not being sent (sender thread blocked)
Symptoms: send() becomes slow or times out.
5. The Sender Thread
Sender Thread Loop
// Pseudocode: Sender thread loop
while (running) {
// 1. Determine which batches are ready
List<ProducerBatch> readyBatches = accumulator.getReadyBatches();
// 2. Group batches by broker (node)
Map<Node, List<ProducerBatch>> batchesByNode = groupByNode(readyBatches);
// 3. Send to each broker
for (Node node : batchesByNode.keySet()) {
List<ProducerBatch> batches = batchesByNode.get(node);
sendBatches(node, batches);
}
// 4. Handle responses
handleResponses();
// 5. Sleep briefly (avoid busy loop)
Thread.sleep(pollTimeoutMs);
}
Sending to Brokers
Sender Thread fetches batches:
Broker 1 (leader for partition 0):
├─ topic-A, partition-0, batch [record1, record2, ...]
└─ topic-B, partition-0, batch [record7, record8, ...]
Broker 2 (leader for partition 1):
└─ topic-A, partition-1, batch [record3, record4, ...]
Broker 3 (leader for partition 2):
└─ topic-A, partition-2, batch [record5, record6, ...]
→ Sender sends one request per broker
→ Each request contains multiple batches (if going to same broker)
In-Flight Requests
max.in.flight.requests.per.connection: Maximum requests sent without acknowledgment.
props.put("max.in.flight.requests.per.connection", 5);
Example: 5 in-flight requests
Time Sender Thread Broker
─────┼──────────────────────────────────┼────────────────
1 │ Send request 1 ───────→ │
2 │ Send request 2 ───────→ │
3 │ Send request 3 ───────→ │
4 │ Send request 4 ───────→ │
5 │ Send request 5 ───────→ │
6 │ Wait for at least one response │ (5 in-flight)
7 │ ←─────── Response for req 1 │
8 │ Send request 6 ───────→ │ (back to 5 in-flight)
Impact:
- Higher value → More throughput (pipelining)
- Lower value → Less throughput, better ordering guarantee (with retries)
In-Flight Requests and Ordering
Critical Insight: max.in.flight.requests.per.connection affects ordering when retries are enabled.
Scenario: Ordering Violation with Retries
max.in.flight.requests.per.connection = 5
retries = 10
Time Sender Thread Broker
─────┼──────────────────────────────────┼────────────────
1 │ Send batch 1 (seq=0) ───────→ │
2 │ Send batch 2 (seq=1) ───────→ │
3 │ Send batch 3 (seq=2) ───────→ │
4 │ │ Batch 1 fails (transient error)
5 │ ←─────── Error for batch 1 │
6 │ ←─────── Success for batch 2 │ ✅ Batch 2 committed (offset 100)
7 │ ←─────── Success for batch 3 │ ✅ Batch 3 committed (offset 101)
8 │ Retry batch 1 ───────→ │ ✅ Batch 1 committed (offset 102)
Result: Batch order = 2, 3, 1 (WRONG!)
Problem: Batch 2 and 3 succeeded while batch 1 was retrying, violating ordering.
Solution 1: Idempotent Producer (Recommended)
props.put("enable.idempotence", true);
props.put("max.in.flight.requests.per.connection", 5); // Safe with idempotence
How it works:
- Producer assigns sequence numbers to batches
- Broker detects out-of-order batches and rejects them
- Producer retries until ordering is restored
With idempotence:
Time Sender Thread Broker
─────┼──────────────────────────────────┼────────────────
1 │ Send batch 1 (seq=0) ───────→ │
2 │ Send batch 2 (seq=1) ───────→ │
3 │ Send batch 3 (seq=2) ───────→ │
4 │ │ Batch 1 fails
5 │ ←─────── Error for batch 1 │
6 │ ←─────── Batch 2 arrives │ ❌ Rejected (seq=1, expected seq=0)
7 │ ←─────── Batch 3 arrives │ ❌ Rejected (seq=2, expected seq=0)
8 │ Retry batch 1 ───────→ │ ✅ Batch 1 committed (seq=0)
9 │ Retry batch 2 ───────→ │ ✅ Batch 2 committed (seq=1)
10 │ Retry batch 3 ───────→ │ ✅ Batch 3 committed (seq=2)
Result: Batch order = 1, 2, 3 (CORRECT!)
Solution 2: Limit In-Flight Requests (Legacy)
props.put("enable.idempotence", false);
props.put("max.in.flight.requests.per.connection", 1); // Only 1 at a time
How it works:
- Only one request in-flight at a time
- No subsequent batches sent until previous succeeds
- Guarantees ordering but severely limits throughput
Tradeoff:
- ✅ Ordering preserved
- ❌ Very low throughput (no pipelining)
- ❌ Not recommended for modern Kafka
Recommendation
Always use idempotent producer:
props.put("enable.idempotence", true); // Default in Kafka 3.0+
props.put("max.in.flight.requests.per.connection", 5); // Default
Benefits:
- ✅ Ordering guaranteed
- ✅ High throughput (pipelining)
- ✅ No duplicates on retries
- ✅ Best of both worlds
Only use max.in.flight.requests.per.connection = 1 if:
- You're on a very old Kafka version (< 0.11) without idempotence
- You absolutely cannot enable idempotence (rare)
6. Metadata Management
What is Metadata?
Metadata includes:
- List of brokers
- Topics and partition count
- Leader broker for each partition
- ISR (In-Sync Replicas) for each partition
Why needed: Producer must know which broker is the leader for each partition.
Metadata Refresh
Producer starts
│
├─ Fetch metadata from broker
│ (connect to any broker from bootstrap.servers)
│
└─ Cache metadata in memory
Application sends record to partition 0
│
└─ Lookup leader for partition 0 in cache
└─ Leader: Broker 2
└─ Send to Broker 2
Metadata becomes stale (leader changed)
│
└─ Producer receives NOT_LEADER_FOR_PARTITION
└─ Refresh metadata
└─ Retry with new leader
Metadata Refresh Triggers
- On producer creation — Initial metadata fetch
- Topic doesn't exist in cache — Fetch on first send
- Error from broker —
NOT_LEADER_FOR_PARTITION,UNKNOWN_TOPIC_OR_PARTITION - Periodic refresh — Every
metadata.max.age.ms(default 5 minutes)
Configuration:
props.put("metadata.max.age.ms", 300000); // 5 minutes
Metadata Request Blocking
// Application thread
producer.send(new ProducerRecord<>("new-topic", "key", "value"));
// If topic metadata not in cache:
// 1. Block up to max.block.ms (default 60 seconds)
// 2. Request metadata from broker
// 3. Wait for response
// 4. Continue send
// If timeout → TimeoutException
Why blocking? Producer needs to know which partition and broker to send to.
7. Request Lifecycle
Detailed Request Flow
┌──────────────────────────────────────────────────────────────────┐
│ REQUEST LIFECYCLE │
├──────────────────────────────────────────────────────────────────┤
│ │
│ 1. Application Thread │
│ ├─ Serialize record │
│ ├─ Determine partition │
│ ├─ Add to RecordAccumulator │
│ └─ Return Future<RecordMetadata> │
│ │
│ 2. Sender Thread (when batch ready) │
│ ├─ Fetch batch from accumulator │
│ ├─ Group by destination broker │
│ ├─ Create ProduceRequest │
│ │ - Include PID, Epoch (if idempotent) │
│ │ - Include sequence number (if idempotent) │
│ │ - Include transaction ID (if transactional) │
│ └─ Send over network │
│ │
│ 3. Broker Processes Request │
│ ├─ Validate request (permissions, version, etc.) │
│ ├─ Check for duplicates (if idempotent) │
│ ├─ Write to partition log │
│ ├─ Replicate to followers (if acks=all) │
│ └─ Send ProduceResponse │
│ │
│ 4. Sender Thread Receives Response │
│ ├─ Success? Update metadata (offset, timestamp) │
│ ├─ Error? Determine if retriable │
│ │ - Retriable → Retry (back to accumulator) │
│ │ - Non-retriable → Fail │
│ └─ Invoke callback │
│ │
│ 5. Callback Execution (in Sender Thread) │
│ └─ Application-provided callback invoked │
│ │
└──────────────────────────────────────────────────────────────────┘
ProduceRequest Structure
ProduceRequest {
acks: 1 (or 0, -1/all)
timeout: 30000 ms
transactionalId: "my-tx-id" (if transactional)
topicData: [
{
topic: "orders",
partitions: [
{
partition: 0,
records: [
{
offset: (assigned by broker)
timestamp: 1674567890123,
key: "user-123",
value: "order-data",
headers: [...],
// If idempotent:
producerId: 12345,
producerEpoch: 0,
sequence: 42
},
...
]
}
]
}
]
}
ProduceResponse Structure
ProduceResponse {
responses: [
{
topic: "orders",
partitions: [
{
partition: 0,
errorCode: 0 (NONE = success),
baseOffset: 1234567, // Offset of first record
logAppendTime: 1674567890456,
logStartOffset: 0
}
]
}
]
}
8. Compression Internals
Where Compression Happens
Application Thread Sender Thread
│ │
├─ send(record1) │
├─ send(record2) │
├─ send(record3) │
│ │
│ Records added to batch (uncompressed)
│ │
│ ├─ Fetch batch
│ ├─ Compress entire batch
│ └─ Send compressed bytes
Key Point: Compression happens per batch, not per record.
Compression Ratio
Batch (uncompressed):
record1: 100 bytes
record2: 100 bytes
record3: 100 bytes
Total: 300 bytes
Compression type: lz4
Batch (compressed): 150 bytes
Compression ratio: 2:1 (50% reduction)
Benefits:
- Reduced network bandwidth
- Reduced broker disk usage
- Higher effective throughput
Cost:
- CPU overhead on producer
- CPU overhead on consumer (decompression)
9. Flow Control and Backpressure
When Producer is Slower Than Broker
Application sends → RecordAccumulator → Sender → Broker
(buffer)
Buffer usage: 10% (plenty of space)
Producer: Happy ✅
When Producer is Faster Than Broker
Application sends FAST → RecordAccumulator → Sender (slow) → Broker (slow)
(buffer fills up)
Buffer usage: 90%
Buffer usage: 95%
Buffer usage: 99%
Buffer usage: 100% (FULL!)
Application: send() blocks ⏸️
Backpressure: Producer slows down automatically when buffer fills.
Handling Backpressure
Option 1: Block (Default)
props.put("max.block.ms", 60000); // Block up to 60 seconds
producer.send(record); // Blocks if buffer full
Effect: Application slows down naturally.
Option 2: Fail Fast
props.put("max.block.ms", 0); // Don't block
try {
producer.send(record);
} catch (org.apache.kafka.common.errors.TimeoutException e) {
// Buffer full — reject send
logger.warn("Buffer full, dropping record");
}
Effect: Application handles overflow explicitly.
Note on Exception Types:
When the buffer is exhausted, you may see:
org.apache.kafka.common.errors.TimeoutException— Failed to allocate memory withinmax.block.msorg.apache.kafka.common.errors.BufferExhaustedException— Explicit indication buffer is full
Both indicate the same root cause: producer buffer memory is exhausted.
Common Causes:
- Sending faster than broker can consume
- Network issues or broker slowness
buffer.memorytoo small for workload- Sender thread blocked or slow
Solutions:
- Increase
buffer.memory - Reduce send rate (implement backpressure)
- Check broker/network health
- Tune
batch.sizeandlinger.msfor faster flushing
Option 3: Increase Buffer
props.put("buffer.memory", 67108864); // 64 MB (was 32 MB)
Effect: More headroom, but doesn't solve root cause.
10. Why Configurations Matter
Understanding Through Internals
Now you understand why these configs exist:
batch.size
Why it matters: Determines when RecordAccumulator considers batch full.
- Small: More frequent sends, lower latency, less throughput
- Large: Fewer sends, higher latency, more throughput
linger.ms
Why it matters: How long Sender waits to fill batch before sending.
- 0ms: Send as soon as any record arrives (low latency)
- >0ms: Wait to batch more records (high throughput)
buffer.memory
Why it matters: Total memory for RecordAccumulator.
- Small: Buffer fills quickly → Backpressure
- Large: More headroom, but uses more memory
max.in.flight.requests.per.connection
Why it matters: How many requests Sender can send without waiting for responses.
- 1: Strict ordering (but slow)
- 5: Good throughput with ordering (if idempotent)
- >5: Max throughput (may reorder on retry)
compression.type
Why it matters: Sender compresses batches before sending.
- none: No CPU overhead, more network
- lz4/snappy: Balanced
- gzip/zstd: Max compression, high CPU
11. Internal Metrics
Key Internal Metrics to Monitor
// Buffer usage
kafka.producer:type=producer-metrics,client-id=<id>,name=buffer-available-bytes
→ How much buffer space is free
kafka.producer:type=producer-metrics,client-id=<id>,name=buffer-exhausted-rate
→ How often buffer fills up (backpressure)
// Batching
kafka.producer:type=producer-metrics,client-id=<id>,name=batch-size-avg
→ Average batch size
kafka.producer:type=producer-metrics,client-id=<id>,name=records-per-request-avg
→ How many records per request
// Sender thread
kafka.producer:type=producer-metrics,client-id=<id>,name=request-rate
→ Requests per second
kafka.producer:type=producer-metrics,client-id=<id>,name=request-latency-avg
→ Average request latency
Summary
Key Takeaways
✅ Two threads: Application thread (adds to buffer), Sender thread (network I/O)
✅ RecordAccumulator: In-memory buffer for batching
✅ Batching: Records grouped by (topic, partition) before sending
✅ Batch ready when: Full (batch.size), linger expires (linger.ms), or buffer full
✅ Sender thread: Fetches batches, sends to brokers, handles responses
✅ Metadata: Cached, refreshed on errors or periodically
✅ In-flight requests: Multiple requests can be outstanding (max.in.flight.requests)
✅ Compression: Happens per batch in Sender thread
✅ Backpressure: send() blocks when buffer full (max.block.ms)
✅ Callbacks: Executed in Sender thread (keep them fast)
Mental Model
Application sends → Serializer → Partitioner → RecordAccumulator (buffer)
↓
[Batch by partition]
↓
Sender Thread (background)
↓
[Compress batches]
↓
[Send to brokers]
↓
[Handle responses]
↓
[Invoke callbacks]
Why This Matters
Understanding internals helps you:
- Debug performance issues
- Tune configurations intelligently
- Predict behavior under load
- Troubleshoot mysterious failures
- Design efficient producer applications
What's Next
Module 8: Performance & Tuning
Now that you understand internals, learn how to optimizeModule 9.1: Practical Producer Patterns
Apply knowledge to real-world scenarios
Previous Module: ← 6.1 - Error Handling & Failure Scenarios
Next Module: 8.1 - Batching & Buffering →