Producer Workflow #

In the Apache Kafka ecosystem, the producer acts as the main gateway for inserting data into the broker cluster. The producer’s role isn’t as simple as sending data through an ordinary TCP socket connection. Behind the seemingly simple .send() function call in our applications, there’s a very complex, asynchronous, and highly optimized internal architecture built to achieve massive throughput with minimal latency. Kafka producers are designed by decoupling completely the main application thread from the network I/O thread. Understanding how the producer’s internal components work together is a key requirement for every system architect to design resilient, efficient message-sending applications.

Message Delivery Lifecycle: Step by Step #

When our application calls producer.send(record), the message doesn’t immediately fly over the network. The message must pass through a series of internal processing steps inside the client JVM memory.

Here’s the message journey from our application code to the broker:

1. Cluster Metadata Fetching #

Before a message can be processed, the producer must know the current physical structure of the Kafka cluster: which brokers are alive, what topics are available, what partitions those topics have, and which broker acts as the Leader for each partition.

  • Mechanism: If metadata isn’t already in the client’s local memory (or has expired), the producer performs an asynchronous blocking operation to request the latest metadata from one of the active brokers (bootstrap servers). Once obtained, the metadata is stored in the client’s local memory cache.

2. Data Serialization #

Kafka is a raw byte storage system. It doesn’t care about our application’s object structures, classes, or data types.

  • Mechanism: The Serializer component converts the message key and value objects into byte arrays (byte[]). Kafka provides built-in serializers for basic data types (String, Integer, Double, ByteArray), but we can use custom formats like Avro or Protobuf.

3. Partitioning #

After data is converted to bytes, the producer must decide which partition the message should be sent to. This task is delegated to the Partitioner.

  • Mechanism: If the message has a key, the default partitioner uses the Murmur2 hash algorithm to consistently bind that key to a specific partition number. If the message has no key, Kafka uses the Sticky Partitioner to distribute data in balanced batches.

4. Memory Buffering (Record Accumulator) #

After the destination partition is determined, the message enters the heart of the Kafka producer architecture: the Record Accumulator. This is where thread division happens. The main application thread’s job ends here and it can immediately return to business calculations, while the message is temporarily stored in buffer memory.

  • Mechanism: The Record Accumulator groups messages by their destination topic and partition. Each partition has a batch queue (double-ended queue of ProducerBatch). New messages are appended to the end of the current active batch.

5. Network Sender Thread #

The final component is the Sender Thread, a background daemon thread responsible for the actual network I/O operations.

  • Mechanism: This thread continuously monitors the Record Accumulator. When a batch is deemed ready to send (because its size is full or its wait time has elapsed), the Sender Thread takes that batch, assembles it into a single network write request (Produce Request), and sends it asynchronously to the partition leader broker using the Java NIO Selector (epoll or kqueue at the OS level).

Producer Internal Thread Architecture #

Visualizing the internal architecture of a Kafka producer shows the strict separation between the application sending thread (App Thread) and the network sending thread (Sender Thread):

flowchart TD
    subgraph App_Execution["1. Main Application Thread (App Thread)"]
        direction TB
        SendCall["producer.send(record)"] --> Metadata["1. Metadata Fetcher"]
        Metadata --> Serializer["2. Serializer (Object -> byte[])"]
        Serializer --> Partitioner["3. Partitioner (Choose Partition)"]
    end

    subgraph Memory_Accumulator["2. Buffer (Record Accumulator)"]
        direction TB
        subgraph Partition_Queues["Batch Queues Per Partition"]
            P0_Q["Partition 0 Queue <br> [Batch A] -> [Batch B]"]
            P1_Q["Partition 1 Queue <br> [Batch C]"]
        end
        BP["Buffer Pool Manager (RAM)"]
    end

    subgraph Network_Execution["3. Network Thread (Sender Thread)"]
        direction TB
        Sender["Sender I/O Thread"] --> Selector["Java NIO Selector"]
    end

    Partitioner -->|"Send to Buffer"| Partition_Queues
    BP -.->|"Allocate Memory"| Partition_Queues
    Partition_Queues -.->|"Take Ready Batch"| Sender
    Selector -->|"Send Packets to Broker via Socket"| Broker["Kafka Broker Leader"]

    style App_Execution stroke:#e5e7eb
    style Memory_Accumulator stroke:#e5e7eb
    style Network_Execution stroke:#e5e7eb
    style P0_Q stroke:#0288d1,stroke-width:2px
    style P1_Q stroke:#2e7d32,stroke-width:2px

Record Accumulator Data Structure & Concurrency Control #

At the Java code level, the Record Accumulator is managed using data structures highly optimized for concurrent multi-thread access.

The per-partition queues are stored in a thread-safe map:

// Internal Record Accumulator data structure
ConcurrentMap<TopicPartition, ArrayDeque<ProducerBatch>> batches;

Why Is This Design So Fast? #

  1. Thread-Safe Map: ConcurrentMap allows multiple application threads to call .send() in parallel for different partitions without blocking each other.
  2. Fine-Grained Locking: When multiple threads try to write messages to the same partition, Kafka doesn’t lock the entire Record Accumulator. Locking is scoped only to the ArrayDeque queue level of that specific partition.
  3. ProducerBatch: Instead of copying messages one by one, new messages are directly transitioned into the physical byte array inside the active ProducerBatch buffer using a fast memory copy system call (System.arraycopy).

Sender Thread Work Cycle (Sender Thread Mechanics) #

The Sender Thread is a non-blocking I/O engine executing the Java NIO Selector Loop. This thread continuously runs the following evaluation cycle:

1. Accumulator Scan (Ready Check) #

The Sender Thread checks all partition queues to determine which batches are “ready to send”. A batch is declared ready if:

  • The batch size has fully reached the batch.size parameter.
  • The batch has been held in the accumulator for longer than the linger.ms parameter.
  • A forced flush call happens (producer.flush()).

2. Node Grouping #

Messages in the accumulator are organized by partition. However, TCP networking operates per server (Broker Node).

  • Consolidation: The Sender Thread performs regrouping. If there are batches for TopicA-Partition0 and TopicB-Partition2 that happen to have the same leader broker (e.g., Broker 2), the Sender Thread combines both batches into a single ClientRequest.
  • Efficiency: This step cuts TCP header overhead and optimizes network packet size.

3. In-Flight Request Control #

Before sending requests to the socket, the Sender Thread verifies the in-flight request limit:

  • The max.in.flight.requests.per.connection property (default: 5) controls the number of pending requests (not yet ACKed) that a producer may send to one broker simultaneously.
  • If this limit is reached, the Sender Thread holds new requests for that broker to prevent network congestion.

Buffer Pool Memory Allocation Timeline Simulation #

Let’s simulate how BufferPool memory is allocated and returned while an asynchronous producer runs with batch.size=16KB:

  • Second 0.000: The Application Thread calls .send() for a 10 KB message on Partition 0.
    • The Record Accumulator sees there’s no active batch for Partition 0 yet.
    • The Accumulator requests 16 KB of memory from the Buffer Pool.
    • The Buffer Pool provides one empty ByteBuffer (16 KB). The remaining Buffer Pool RAM is 32 MB - 16 KB.
    • The 10 KB message is written to that ByteBuffer. The remaining free space in that batch is 6 KB.
  • Second 0.002: The Application Thread sends a second 5 KB message to Partition 0.
    • The Accumulator sees Partition 0’s active batch still has free space (6 KB > 5 KB).
    • The 5 KB message is directly written to the same ByteBuffer. Remaining free space is 1 KB.
  • Second 0.004: The Application Thread sends a third 5 KB message to Partition 0.
    • The Accumulator detects the remaining free space isn’t enough (1 KB < 5 KB).
    • The first batch (15 KB) is declared CLOSED and ready to be taken by the Sender Thread.
    • The Accumulator requests a new 16 KB ByteBuffer from the Buffer Pool to hold that 5 KB message.
  • Second 0.005: The Sender Thread takes the first closed batch, sends it to the broker, and receives a successful ACK.
    • The first batch’s ByteBuffer memory is completely cleaned.
    • That memory is returned to the Buffer Pool. The Buffer Pool RAM increases back by 16 KB.

This memory recycling pattern eliminates dynamic byte array object creation that could trigger Garbage Collection pauses.

Comparison: Asynchronous vs Synchronous Delivery #

As application developers, we have full control to execute data delivery asynchronously (using callbacks) or synchronously (blocking the thread to wait for results).

1. Synchronous Scenario (ANTI-PATTERN for High Throughput) #

Synchronous delivery is done by calling the .get() function directly on the Future object returned by .send().

// ANTI-PATTERN: Blocking the application thread, crippling throughput
public class SynchronousProducer {
    public void sendData(KafkaProducer<String, String> producer) {
        for (int i = 0; i < 10000; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", "Key-" + i, "Data-" + i);
            try {
                // Calling .get() forces the thread to wait for an ACK from the broker before sending the next message
                RecordMetadata metadata = producer.send(record).get(); // ✗ Throughput plummets!
                System.out.printf("Message successfully written to offset %d%n", metadata.offset());
            } catch (Exception e) {
                System.err.println("Failed to send data: " + e.getMessage());
            }
        }
    }
}
  • Bad Impact: Throughput is completely destroyed because the delivery cycle is limited by network RTT (Round Trip Time) latency. If the network RTT is 10 ms, our maximum throughput is only 100 messages per second per producer thread.

2. Asynchronous Scenario (CORRECT & Recommended) #

True asynchronous delivery uses callbacks to handle results without ever holding up the main business thread execution.

// CORRECT: Using asynchronous Callbacks, optimal throughput
public class AsynchronousProducer {
    public void sendData(KafkaProducer<String, String> producer) {
        for (int i = 0; i < 10000; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>("orders", "Key-" + i, "Data-" + i);
            
            // Non-blocking delivery, immediately continue to the next iteration
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        // ✓ Handle the error on a separate thread (Sender Thread)
                        System.err.printf("Failed to send message: %s%n", exception.getMessage());
                    } else {
                        // ✓ Successfully recorded data
                        System.out.printf("Message successfully written to partition %d, offset %d%n", 
                            metadata.partition(), metadata.offset());
                    }
                }
            });
        }
    }
}

Memory Safety and Backpressure Mechanism #

What happens if our producer generates data at 100 MB/second, while our network can only deliver data to the broker at 20 MB/second?

This is where Kafka’s Backpressure defense system kicks in:

  1. New messages keep entering the Record Accumulator.
  2. Because data isn’t getting sent, the Buffer Pool runs out of empty ByteBuffers. The 32 MB memory is completely full.
  3. When the next .send() call is invoked by the application, that function doesn’t allocate new RAM (which could trigger OOM).
  4. The .send() function is held synchronously (blocked) for the duration of the max.block.ms parameter (default: 60,000 ms / 1 minute).
  5. Our application thread naturally slows down following network speed. This is a very safe backpressure mechanism.
  6. If within 1 minute the broker hasn’t recovered or the buffer memory hasn’t been freed, the producer eventually gives up and throws the TimeoutException.

Viewing JMX Buffer Monitoring Metrics #

To detect bottleneck or backpressure symptoms in producers, we must monitor the following JMX metrics:

# Measuring the average time application threads are blocked waiting for buffer pool memory
kafka.producer:type=producer-metrics,client-id=[clientId],name=bufferpool-wait-time-ns

# Measuring the average time message records are held in the Record Accumulator before being sent
kafka.producer:type=producer-metrics,client-id=[clientId],name=record-queue-time-avg

Automatic Cluster Metadata Update Mechanism #

Cluster metadata is the producer’s road map. Without accurate metadata, the producer won’t know which broker acts as the partition Leader to deliver our messages. Therefore, Kafka clients have an internal mechanism to always update this metadata automatically and asynchronously.

1. Periodic Refresh #

The producer performs periodic background metadata updates every time the wait period reaches the metadata.max.age.ms parameter (default set to 300,000 ms or 5 minutes). This is done to ensure the producer detects if new partitions are added to a topic, or if new brokers are added to the cluster.

2. Reactive Refresh #

If a failure occurs while sending data (for example, the leader broker suddenly dies), the remaining brokers reject the producer’s write request and throw exceptions like NotLeaderOrFollowerException or KafkaStorageException. When the producer receives this kind of error, it realizes its local metadata map is stale. The producer client immediately marks the local metadata as stale and triggers a reactive metadata refresh as soon as possible before trying to retry sending that message.

Custom Partitioner: Writing Your Own Division Algorithm #

By default, Kafka producers use Murmur2 hashing if we include a key on the message, or the Sticky Partitioner if there’s no key. However, sometimes business needs demand full control over message placement within partitions. To meet those needs, we can implement a custom Partitioner component.

For example, imagine we have an e-commerce order processing system. We want to place all VIP customer transactions on partition 0 to get higher resource processing priority and strict isolation, while regular customer transactions are spread randomly across other partitions.

Here’s a comparison between the wrong way of handling partition division at the application level (anti-pattern) versus the correct way using a custom Partitioner interface:

// ANTI-PATTERN: Determining partitions manually in the main application code
public class OrderService {
    public void sendOrder(KafkaProducer<String, String> producer, Order order) {
        int targetPartition = 0;
        if (order.isVip()) {
            targetPartition = 0; // Sending all VIPs to partition 0 manually
        } else {
            // DON'T DO THIS: Manual hashing logic clutters business code and is hard to maintain
            targetPartition = Math.abs(order.getCustomerId().hashCode()) % 3 + 1;
        }
        
        // ✗ Explicitly specifying the partition number when creating the Record object
        ProducerRecord<String, String> record = new ProducerRecord<>(
            "orders-topic", targetPartition, order.getId(), order.toJson()
        );
        producer.send(record);
    }
}

// CORRECT: Creating a separate Partitioner implementation and registering it via configuration properties
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;

public class CustomerPriorityPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        // Get the total partitions available for this topic on the broker
        int numPartitions = cluster.partitionCountForTopic(topic);
        
        if (key == null) {
            return 0; // Keyless messages directed to partition 0 as a fallback
        }
        
        String keyString = (String) key;
        if (keyString.startsWith("VIP_")) {
            // ✓ Dedicated portion: All VIP customers go to partition 0
            return 0;
        }
        
        // ✓ Standard Murmur2 hashing for regular customers
        // Spread data on partitions other than partition 0 (i.e., partitions 1 to numPartitions - 1)
        int hash = org.apache.kafka.common.utils.Utils.toPositive(
            org.apache.kafka.common.utils.Utils.murmur2(keyBytes)
        );
        return (hash % (numPartitions - 1)) + 1;
    }

    @Override
    public void close() {
        // Cleaning up resources if needed when the producer is shut down
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // Reading additional configuration passed to the producer if any
    }
}

Once our custom partitioner class is complete, we simply register it into the producer configuration properties:

import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

// ✓ Register our custom partitioner class here
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomerPriorityPartitioner.class.getName());

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

By moving the data division logic to a dedicated Partitioner class, our main business code stays clean from Kafka partition infrastructure details, and we can easily swap partition strategies in the future without changing our main order processing code.

Summary #

  • Decoupled Architecture: The producer separates the delivery process between the application thread (App Thread) and the network thread (Sender Thread) asynchronously.
  • Record Accumulator: Acts as a memory buffer container where messages are grouped into batches per partition before being sent.
  • Buffer Pool Manager: A JVM memory optimization recycling fixed-size ByteBuffers of batch.size to avoid Garbage Collection overhead.
  • Node Grouping: The Sender Thread combines several partition batches with the same leader broker into one Produce Request for connection efficiency.
  • Asynchronous Scenario: Always use asynchronous Callbacks on .send() in production environments to avoid network latency bottlenecks.
  • Backpressure: Controlled by the buffer.memory and max.block.ms parameters, safely holding back the application thread rate if buffer capacity is full.
Next: Serialization →
About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact