Kafka Producers

#Introduction

A producer writes records to Kafka. Sending looks like one method call, but behind it is a pipeline of serialization, partitioning, batching, compression, retries and acknowledgements. Understanding that pipeline explains most producer settings and interview questions.

#The Send Pipeline

send() is asynchronous: it adds the record to an in-memory buffer and returns a Future immediately. A background sender thread ships batches to brokers.

#A Plain Java Producer

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>4.1.0</version>
</dependency>

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ACKS_CONFIG, "all");

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

    ProducerRecord<String, String> record =
            new ProducerRecord<>("orders", "ORD-1001", "{\"status\":\"PLACED\"}");

    producer.send(record, (metadata, exception) -> {
        if (exception == null) {
            System.out.printf("Sent to %s-%d at offset %d%n",
                    metadata.topic(), metadata.partition(), metadata.offset());
        } else {
            System.err.println("Send failed: " + exception.getMessage());
        }
    });
}   // close() flushes buffered records before shutting down

Sent to orders-2 at offset 17

Three Ways to Send

Headers

ProducerRecord<String, String> record = new ProducerRecord<>("orders", "ORD-1001", json);
record.headers().add("eventType", "OrderPlaced".getBytes(StandardCharsets.UTF_8));
record.headers().add("traceId", traceId.getBytes(StandardCharsets.UTF_8));

#Serializers

Kafka stores bytes; serializers convert your objects:

#Partitioning

You can plug in a custom Partitioner, but a good key is almost always enough.

#Batching: linger.ms and batch.size

Sending records one by one wastes network round trips. The producer groups records per partition into batches.

A batch is sent when it's full or when linger.ms expires, whichever comes first. Raising linger.ms (e.g. 10–50 ms) and batch.size (e.g. 64–256 KB) improves throughput at the cost of a little latency.

#Compression

props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");   // none, gzip, snappy, lz4, zstd

Compression works on whole batches, so it's more effective with batching enabled. Brokers store the compressed batches as-is and consumers decompress them.

#Acknowledgements (acks)

Default since Kafka 3.0: acks=all.

#Retries and Delivery Timeout

Transient errors (leader changes, network blips, NotEnoughReplicas) are retried automatically:

Non-retriable errors (RecordTooLargeException, serialization errors, authorization failures) fail immediately.

#Idempotence and Ordering

Retries create a risk: the broker writes a batch, the acknowledgement is lost, the producer retries, and the batch is written twice. An idempotent producer prevents this:

• the broker assigns the producer an ID, and the producer numbers each batch per partition;
• the broker discards batches it has already written (same producer ID + sequence number).

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   // default since Kafka 3.0

Idempotence also preserves ordering during retries, provided max.in.flight.requests.per.connection ≤ 5 (the default is 5). Without idempotence, a retried batch could land after a later one.

#Message Size

The default maximum record batch size on brokers is about 1 MB (message.max.bytes), with matching producer (max.request.size) and topic (max.message.bytes) limits. For large payloads (files, images), store them in object storage (S3, Azure Blob) and send a reference in the event — the claim-check pattern.

#Key Producer Settings

#Key Takeaways

• send() is asynchronous; records are batched per partition and sent by a background thread.
• Use one shared, thread-safe producer, and an async send with a callback.
• The key decides the partition; no key means sticky partitioning.
• linger.ms, batch.size and compression tune throughput.
• Defaults since Kafka 3.0: acks=all, idempotence on, effectively unlimited retries bounded by delivery.timeout.ms.
• Idempotence prevents retry duplicates — not application-level duplicates.