Exactly-Once Processing in Apache Kafka with Idempotent Producers and Transactions

The Problem: Duplicates After Retries

A Kafka producer that retries after a network blip can write the same record more than once. If your downstream process cannot tolerate duplicates, a simple retry is not enough. You need exactly-once semantics: each logical record is persisted one time, even when failures and retries occur.

Useful Takeaway

Set enable.idempotence=true on the producer. Kafka assigns the producer a persistent ID and tracks a sequence number per partition, so the broker can discard duplicate retries. For atomic writes across multiple partitions—or for committing a consumer offset together with produced records—use the transactional API with beginTransaction(), sendOffsetsToTransaction(), and commitTransaction(). Consumers must read with isolation.level=read_committed to skip uncommitted data.

Mechanism

Idempotent producers use a producer ID (PID) and a per-partition sequence number. When a broker receives a retry with a sequence number it has already processed, it ignores the duplicate. Transactions extend this to multiple partitions through a transaction coordinator. The coordinator records transaction state in the internal __transaction_state topic and writes commit or abort markers so consumers can see a consistent boundary.

Version assumption: this behavior applies to Kafka 2.5 and later. enable.idempotence=true is the default since Kafka 3.0, but setting it explicitly is safer when you support older clients or brokers.

Worked Example: Transactional Producer in Java

Run this in a Java application using the Kafka client library. The application needs permission to read from the source topic, write to the sink topic, and describe consumer group offsets.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "");
props.put(ProducerConfig.TRANSACTION_TIMEOUT_MS_CONFIG, "");

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

try {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    producer.beginTransaction();

    for (ConsumerRecord<String, String> record : records) {
        // process record
        producer.send(new ProducerRecord<>("", record.key(), record.value()));
    }

    producer.sendOffsetsToTransaction(consumerOffsets, "");
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

Replace placeholders with your cluster address, a unique transactional ID for this producer instance, the sink topic, and the consumer group ID. The transaction timeout must be longer than your expected processing time; otherwise the broker aborts the transaction.

Expected check: after a forced producer restart, consume the sink topic with isolation.level=read_committed and confirm each logical record appears once. Risk: reusing the same transactional.id across concurrent producer instances causes ProducerFencedException and aborts the older producer.

Consumer-Side Requirement

Exactly-once delivery to the application requires the consumer to filter out uncommitted messages. Set this property in the consumer configuration:

props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

With the default read_uncommitted, a consumer can read records from a transaction that later aborts, which breaks exactly-once processing.

Limits and Common Mistakes

• Transactions cannot span multiple Kafka clusters. Cross-cluster replication requires a separate design, such as MirrorMaker with idempotent consumers.
• Exactly-once applies only inside Kafka. Writing to an external database or API in the same transaction is not atomic; use an idempotent sink or an outbox pattern.
• Forgetting sendOffsetsToTransaction() commits the consumer offset outside the transaction, which can produce duplicates after a failure.
• Sharing one transactional.id between concurrent producer instances fences the older instance.
• Processing that exceeds transaction.timeout.ms is aborted by the broker.

Verification

1. Start a transactional producer writing to a sink topic.
2. Restart the producer or broker mid-flow to simulate failure.
3. Consume the sink topic with isolation.level=read_committed and check for duplicate keys or sequence numbers.
4. Inspect broker logs for transaction markers or the __transaction_state internal topic for commit and abort markers.
5. Measure throughput impact with kafka-producer-perf-test --transactional-id .... Run this on a client machine with the Kafka binaries installed; the test generates load, so avoid running it against a production cluster during peak traffic.

Rollback

If an exception occurs before commit, call abortTransaction(). This discards the uncommitted writes and writes an abort marker so read_committed consumers never see the partial batch.

Practical Result Check

Compare the number of records read from the source topic against the number committed to the sink topic after a restart. The counts should match, with no duplicate keys in the sink.