Configuring Apache Kafka Producer Idempotence for Exactly‑Once Delivery

Desired Outcome

Enable producer idempotence so that automatic retries do not create duplicate records. When idempotence is active, each message sent by the producer is written exactly once to its target partition, even if the broker temporarily fails and the producer retries.

Prerequisites

• Kafka broker version 0.11.0 or later (idempotent producer was introduced in 0.11).
• Producer client library version that matches or is newer than the broker version.
• The producer must be configured with acks=all (or -1) and retries > 0.
• If you intend to use transactions, provide a unique transactional.id; otherwise, idempotence can be enabled without transactions.

Procedure

Follow these steps to create an idempotent producer in a Java application.

1. Set the required properties. The key properties are:
• enable.idempotence=true
• acks=all
• retries=Integer.MAX_VALUE (or a sufficiently high value, e.g., Integer.MAX_VALUE).
• max.in.flight.requests.per.connection=5 (the default; do not increase beyond 5 when idempotence is enabled).
• Optional: transactional.id=my‑txn‑id if you also need exactly‑once semantics across multiple partitions.
2. Create the producer instance.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
// If using transactions:
// props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-txn-id");

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

3. Send records as usual. No additional logic is needed for deduplication.

ProducerRecord<String, String> record = new ProducerRecord<>("orders", "order-123", "{"item":"widget","qty":5}");
producer.send(record);
// Optionally flush or close when done.
producer.flush();

Expected Checks

After starting the producer, verify that idempotence is active and that no duplicates appear.

1. Inspect producer configuration. Retrieve the effective config and confirm the key flags.

Map<String, Object> configs = producer.configs();
System.out.println("enable.idempotence: " + configs.get(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG));
System.out.println("acks: " + configs.get(ProducerConfig.ACKS_CONFIG));
System.out.println("retries: " + configs.get(ProducerConfig.RETRIES_CONFIG));

2. Check broker logs for idempotent initialization. On the broker, look for lines similar to:

INFO [ProducerIdempotentManager brokerId=0] Registered producer with producerId=5 and epoch=0

3. Monitor producer metrics. JMX metrics record-retry-rate and record-send-rate should show retries only when transient errors occur, and the retry count should not increase the number of records committed.
4. Consume and validate. Use a fresh consumer group to read the topic and ensure each key appears exactly once.

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "verify-idempotence");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("orders"));

Map<String, Integer> counts = new HashMap<>();
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(30));
for (ConsumerRecord<String, String> r : records) {
    counts.merge(r.key(), 1, Integer::sum);
}

for (Map.Entry<String, Integer> e : counts.entrySet()) {
    if (e.getValue() > 1) {
        System.err.println("Duplicate key found: " + e.getKey() + " count=" + e.getValue());
    } else {
        System.out.println("Key " + e.getKey() + " appears exactly once.");
    }
}

Recovery Options

If idempotence does not behave as expected, follow these steps to diagnose and recover.

1. Verify broker version. Run kafka-broker-api-versions.sh --bootstrap-server kafka-broker:9092 and ensure the broker reports version 0.11.0 or higher.
2. Check producer client version. Confirm that the client library matches the broker (e.g., both 3.5.x).
3. Ensure acks=all is set. Any value lower than all disables idempotence; correct the property and restart the producer.
4. Review max.in.flight.requests.per.connection. If you increased this beyond 5, reduce it back to 5 or less to preserve ordering guarantees.
5. Transactional ID conflicts. When using transactions, each transactional.id must be unique across producer instances. Duplicate IDs cause the broker to reject the producer; resolve by assigning a distinct ID.
6. Fallback to non‑idempotent mode. As a temporary measure, set enable.idempotence=false and rely on exactly‑once semantics via Kafka Streams transactions or consumer‑side deduplication. Remember that this may allow duplicates on retries.
7. Monitor for errors. Look for ProducerFencedException or OutOfOrderSequenceException in logs; these indicate a breach of idempotence assumptions and require restarting the producer with a new producerId (i.e., create a new producer instance).

Limitations

• Idempotence only guarantees exactly‑once delivery within a single partition. Cross‑partition exactly‑once requires transactions.
• Enabling idempotence adds a small overhead due to producer ID and sequence number tracking.
• The feature does not protect against logical duplicates produced by the application itself (e.g., sending the same record twice).

By following the steps above, you can enable and validate Kafka producer idempotence, achieving reliable exactly‑once semantics for your streaming workloads.