Kafka Exactly-Once Semantics & Transactions

Kafka supports exactly-once semantics (EOS) through a combination of idempotent producers and transactional APIs. Together they ensure that a message is produced to Kafka exactly once even in the presence of retries, and that a read-process-write operation either fully commits or fully aborts atomically.

Kafka

Delivery Guarantees
Idempotent Producer
Transactional Producer
Read-Process-Write Pattern
Limitations

Delivery Guarantees

Guarantee	How	Risk
At-most-once	Fire and forget: `acks=0`, no retries, auto-commit before processing	Message loss is possible
At-least-once	`acks=all`, retries enabled, manual commit after processing	Duplicates possible on retry
Exactly-once	Idempotent producer + transactions + `read_committed` consumer	Higher latency and complexity

For the vast majority of applications, at-least-once with idempotent consumer logic is the right choice. Reserve exactly-once for financial, billing, or deduplication-sensitive workloads.

Idempotent Producer

An idempotent producer assigns a sequence number to every message. The broker deduplicates retries within a single producer session, ensuring each message is written exactly once per partition.

Enabled by setting enable.idempotence=true (default in Kafka 3+).
Requires acks=all and max.in.flight.requests.per.connection ≤ 5.
Idempotence covers one producer session, if the producer restarts, duplicates can still occur without transactions.

Java

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());

// Enable idempotence — deduplicates retries within a session
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);  // sets acks=all automatically
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("orders", "key1", "order data"));
producer.flush();
producer.close();

Transactional Producer

A transactional producer groups multiple sends (potentially to different topics/partitions) into an atomic unit. Either all writes commit or all are aborted: consumers configured with isolation.level=read_committed will never see aborted records.

Java

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.ENABLE_IDEMPOTENCE_CONFIG, true);

// Unique ID that survives producer restarts — required for transactions
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions(); // register with transaction coordinator

try {
    producer.beginTransaction();

    producer.send(new ProducerRecord<>("orders", "order-1", "order data"));
    producer.send(new ProducerRecord<>("order-events", "order-1", "ORDER_CREATED"));

    producer.commitTransaction(); // both writes become visible atomically
} catch (ProducerFencedException | OutOfOrderSequenceException e) {
    // Fatal — must close producer
    producer.close();
} catch (KafkaException e) {
    producer.abortTransaction(); // roll back — both writes are discarded
}

Each transactional.id must be unique per producer instance. If a new producer starts with the same ID, the old producer is fenced and any open transaction is aborted.

Read-Process-Write Pattern

The canonical exactly-once pattern reads from an input topic, processes the records, and writes results along with consumer offsets atomically in a single transaction.

Java

// Consumer — must use read_committed to skip aborted transaction messages
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "eos-processor");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(List.of("input-topic"));

// Producer — transactional
Properties producerProps = new Properties();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "eos-producer-1");

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

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    try {
        producer.beginTransaction();

        for (ConsumerRecord<String, String> record : records) {
            String result = transform(record.value()); // your processing logic
            producer.send(new ProducerRecord<>("output-topic", record.key(), result));
        }

        // Atomically commit offsets as part of the transaction
        Map<TopicPartition, OffsetAndMetadata> offsets = currentOffsets(records);
        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

        producer.commitTransaction(); // all writes + offset commit are atomic
    } catch (KafkaException e) {
        producer.abortTransaction();
        consumer.seekToBeginning(consumer.assignment()); // reprocess from committed offset
    }
}

sendOffsetsToTransaction() commits the consumer offsets as part of the producer transaction. This means the offset is only committed if the transaction commits, achieving exactly-once end-to-end.

Limitations

Performance overhead: transactions add latency. The transaction coordinator must persist the transaction state before commit. transaction.timeout.ms defaults to 60 seconds.
Single Kafka cluster only: exactly-once across multiple Kafka clusters is not supported.
No cross-system EOS: Kafka transactions do not extend to external databases or systems. If your processing involves a database write, you need additional techniques (outbox pattern, saga).
Zombie fencing: a stale producer with the same transactional.id is fenced when a new producer initialises. Plan for this in deployments: use unique IDs per consumer partition for fan-out patterns.
Kafka Streams EOS: if using Kafka Streams, set processing.guarantee=exactly_once_v2 instead of managing transactions manually.

Contents

Delivery Guarantees

Idempotent Producer

Transactional Producer

Read-Process-Write Pattern

Limitations

Related articles