Kafka - Producers

A Kafka Producer is a client application that publishes (writes) records to Kafka topics. It handles partitioning, batching, compression, retries, and acknowledgment: all configurable for your reliability and throughput requirements.

Kafka

Basic Producer Example
Acknowledgment (acks)
Partitioning
Batching and Compression
Idempotent and Transactional Producers
Key Configuration Summary

Basic Producer Example

Add the Kafka clients dependency to your project (use the latest version; the examples here follow the defaults of Kafka 4.x):

XML

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

Java

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

public class SimpleProducer {

    public static void main(String[] args) {
        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());

        // close() at the end of the try block waits for messages still being sent
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>("my-topic", "user-123", "event-payload");

            // send() returns immediately; the callback runs once the broker has answered
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("Send failed: " + exception.getMessage());
                } else {
                    System.out.printf("Sent to partition=%d offset=%d%n",
                            metadata.partition(), metadata.offset());
                }
            });
        }
    }
}

The other examples in this article show only the settings or calls that change, inside this same main method.

Acknowledgment (acks)

acks controls how many broker acknowledgments the producer waits for before considering the request complete. It directly affects durability vs throughput.

acks value	Meaning	Durability	Throughput
`0`	The producer does not wait for any confirmation	Lowest (data loss possible)	Highest
`1`	The partition leader confirms once it has written the message	Medium (loss if leader crashes before replication)	High
`all` or `-1` (default since Kafka 3.0)	The leader confirms once every in-sync replica has the message	Highest, when the topic has `min.insync.replicas` of 2 or more	Lower

The producer retries failed sends by itself. Rather than limiting the number of retries, set how long it may keep trying; when the time is up, the callback receives the error:

Java

props.put(ProducerConfig.ACKS_CONFIG, "all");                 // the default, shown for clarity
props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000); // give up after 2 minutes (the default)

Partitioning

Kafka routes messages to partitions by:

With a key: the partition is a hash of the key (murmur2) modulo the number of partitions. The same key always goes to the same partition, so messages for one key stay in order.
No key: the producer fills a batch for one partition, then moves to another (the "sticky" partitioner, since Kafka 2.4). Over time the load is spread evenly, with fuller batches than strict round-robin.
Custom Partitioner: implement org.apache.kafka.clients.producer.Partitioner.

Java

// Message with key — always goes to same partition
new ProducerRecord<>("orders", "customer-456", orderJson);

// Message without key — spread across partitions in batches
new ProducerRecord<>("logs", null, logLine);

// Explicit partition
new ProducerRecord<>("orders", 2, "customer-789", orderJson);

Batching and Compression

The producer buffers records in a per-partition batch to improve throughput. Compression reduces network and storage overhead.

Java

// Batching
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);        // 16 KB batch size
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);            // wait up to 20 ms to fill a batch (default 5 ms)

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

// Buffer (total memory for producer buffering)
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432L); // 32 MB

Idempotent and Transactional Producers

Idempotent producer: when a send is retried, the broker recognises the repeat and stores the message only once, in the right order. It is on by default since Kafka 3.0. It covers retries inside one producer; if your application sends the same message twice, both are stored.

Java

props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);   // the default
// Requires acks=all and max.in.flight.requests.per.connection of 5 or less (both defaults);
// setting acks to 0 or 1 turns idempotence off

Transactional producer: atomic multi-partition / multi-topic writes. All messages in a transaction are either all committed or all aborted.

Java

props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("orders", key, value));
    producer.send(new ProducerRecord<>("audit-log", key, auditValue));
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    // Fatal: another instance took over this transactional.id, or the producer cannot continue
    producer.close();
} catch (KafkaException e) {
    // Anything else: abort, then the transaction can be retried
    producer.abortTransaction();
}

Consumers see these messages only after the commit, and only if they read with isolation.level=read_committed. The default, read_uncommitted, also shows messages from aborted transactions.

Key Configuration Summary

Config	Default	Purpose
`bootstrap.servers`	-	One or more brokers to connect to first, as `host:port` separated by commas; the producer discovers the rest of the cluster from them
`acks`	all	When a write counts as successful (see above)
`delivery.timeout.ms`	120000	How long a send may take, retries included
`batch.size`	16384	Most bytes collected for one partition before a batch is sent
`linger.ms`	5 (0 before Kafka 4.0)	How long to wait for more messages before sending a batch that is not full
`compression.type`	none	Compress batches with `snappy`, `gzip`, `lz4` or `zstd`; fewer bytes on the network and disk for a little CPU
`enable.idempotence`	true (Kafka 3.0+)	No duplicates or reordering from retries

Contents

Basic Producer Example

Acknowledgment (acks)

Partitioning

Batching and Compression

Idempotent and Transactional Producers

Key Configuration Summary