Achieving Exactly-Once Processing in Kafka Consumers

Part of: Exactly-Once vs At-Least-Once Delivery

A Kafka consumer that reads a batch, applies business logic, produces derived records, and commits its offsets is performing four separate operations against a system that can crash between any two of them. Without transactional guarantees, a crash after the produce but before the offset commit causes the consumer to reprocess the same input batch on restart — and re-produce output records that were already written. This runbook implements Kafka’s transactional exactly-once semantics (EOS) for the read-process-write loop, and the idempotent-consumer fallback required the moment a consumer’s side effect leaves the Kafka cluster.

Prerequisites: you understand the distinction between at-least-once delivery and exactly-once processing and why true exactly-once delivery is impossible over an unreliable network. You are running Kafka 2.5 or later (transactional API is stable from 0.11, but read_committed performance and zombie fencing improved substantially by 2.5). You have broker-level transaction.state.log.replication.factor set to at least 3 in production. All code below is copy-pasteable and independently verifiable.

---

How the transaction boundary works

The diagram below shows the full read-process-write cycle: the producer begins a transaction, writes output records, attaches the consumer offsets to the same transaction, and commits — all as one atomic unit that either fully lands or fully vanishes.

---

Step-by-step implementation

Step 1 — Configure a transactional producer

Set a stable transactional.id that is unique per logical producer instance (typically one per partition assignment or per application instance) and call initTransactions() once at startup. The stable ID is what lets the broker fence out a zombie instance of the same producer after a restart.

// Java: transactional producer configuration
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-broker-1:9092,kafka-broker-2:9092");
props.put("transactional.id", "order-processor-partition-0");
props.put("enable.idempotence", "true");
props.put("acks", "all");
props.put("max.in.flight.requests.per.connection", "5");

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

Verify this step independently: restart the process twice in a row and confirm in the broker logs that the producer epoch for order-processor-partition-0 increments on each initTransactions() call — this is the fencing mechanism working correctly.

Step 2 — Implement the read-process-write loop

Wrap the poll-process-produce-commit cycle in a single transaction. sendOffsetsToTransaction attaches the consumer’s offsets to the same atomic unit as the output records, so a crash between the produce and the offset commit cannot happen — both land together or neither does.

// Java: full read-process-write transaction
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(List.of("orders-raw"));

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

    try {
        producer.beginTransaction();
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();

        for (ConsumerRecord<String, String> record : records) {
            String processed = applyBusinessLogic(record.value());
            producer.send(new ProducerRecord<>("orders-processed", record.key(), processed));
            offsets.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            );
        }

        producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        // Fatal — this producer instance has been superseded, must shut down
        producer.close();
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();
    }
}

Python (confluent-kafka) equivalent:

# Python: confluent-kafka transactional read-process-write
from confluent_kafka import Consumer, Producer, TopicPartition, KafkaException

producer_conf = {
    "bootstrap.servers": "kafka-broker-1:9092",
    "transactional.id": "order-processor-partition-0",
}
producer = Producer(producer_conf)
producer.init_transactions()

consumer = Consumer({
    "bootstrap.servers": "kafka-broker-1:9092",
    "group.id": "order-processor",
    "isolation.level": "read_committed",
    "enable.auto.commit": False,
})
consumer.subscribe(["orders-raw"])

while running:
    msgs = consumer.consume(num_messages=100, timeout=0.5)
    if not msgs:
        continue

    try:
        producer.begin_transaction()
        offsets = []
        for msg in msgs:
            processed = apply_business_logic(msg.value())
            producer.produce("orders-processed", key=msg.key(), value=processed)
            offsets.append(TopicPartition(msg.topic(), msg.partition(), msg.offset() + 1))

        producer.send_offsets_to_transaction(offsets, consumer.consumer_group_metadata())
        producer.commit_transaction()
    except KafkaException:
        producer.abort_transaction()

Step 3 — Set read_committed on every downstream consumer

Any consumer reading orders-processed must set isolation.level=read_committed, or it will see records from transactions that were later aborted:

// Java: downstream consumer configuration
Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "kafka-broker-1:9092");
consumerProps.put("group.id", "downstream-billing");
consumerProps.put("isolation.level", "read_committed");
consumerProps.put("enable.auto.commit", "false");

Verify independently: check every consumer group’s effective config with kafka-consumer-groups.sh --describe and confirm isolation.level is not defaulting to read_uncommitted — this is a common deployment drift that silently breaks the exactly-once guarantee for every downstream reader.

Step 4 — Add an idempotent-consumer fallback for external writes

Kafka transactions only cover Kafka-to-Kafka atomicity. The moment a consumer writes to a database, calls a third-party API, or sends an email, that write is outside the transaction boundary. Combine the read-process-write loop with an idempotent consumer dedup check keyed on topic-partition-offset before performing the external side effect:

# Python: dedup-store fallback for consumers that write to Postgres
import psycopg2

conn = psycopg2.connect(dsn="postgresql://user:pass@localhost/orders")

def process_with_dedup(record):
    dedup_key = f"{record.topic()}-{record.partition()}-{record.offset()}"
    with conn:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO processed_offsets (dedup_key) VALUES (%s) "
                "ON CONFLICT (dedup_key) DO NOTHING RETURNING dedup_key",
                (dedup_key,)
            )
            if cur.fetchone() is None:
                return  # already processed — duplicate delivery, discard

            apply_external_side_effect(record.value())
            # commit happens automatically at the end of the `with conn` block

This uses topic-partition-offset as the identifier rather than a business-level ID because it is guaranteed stable and unique per delivery, even across consumer restarts and rebalances. See using Redis SETNX for distributed request deduplication for a Redis-backed variant of this same pattern.

Step 5 — Verify with duplicate injection

Force a redelivery deliberately and confirm the external write happens only once:

# Kill the consumer process mid-batch after it has produced but before offsets commit
kill -9 $(pgrep -f order-processor)

# Restart it — it will re-read the same uncommitted offset range
./start-order-processor.sh

# Confirm the output topic has no duplicate records for the affected offset range
kafka-console-consumer.sh --bootstrap-server kafka-broker-1:9092 \
  --topic orders-processed --isolation-level read_committed \
  --from-beginning --max-messages 20

---

Verification and testing

Confirm transactional state on the broker:

kafka-transactions.sh --bootstrap-server kafka-broker-1:9092 --list
# Look for order-processor-partition-0; state should be Empty or CompleteCommit
# between batches, never a stuck Ongoing state

Confirm consumer isolation level is applied:

kafka-consumer-groups.sh --bootstrap-server kafka-broker-1:9092 \
  --describe --group downstream-billing --verbose

Inject a duplicate at the source and confirm dedup absorbs it:

kafka-console-producer.sh --bootstrap-server kafka-broker-1:9092 \
  --topic orders-raw --property "parse.key=true" --property "key.separator=:" <<'EOF'
order-42:{"amount":100,"currency":"USD"}
order-42:{"amount":100,"currency":"USD"}
EOF
# Verify processed_offsets and downstream side-effect table each show exactly one row for order-42

---

Failure scenarios and debugging

---

SRE checklist

1. kafka_producer_transaction_abort_total — Counter; alert if greater than 1% of transactions over a 10-minute window, indicating coordinator instability or undersized transaction.timeout.ms.
2. producer_fenced_total — Counter; alert on any non-zero value, this always indicates two live producer instances sharing a transactional.id.
3. consumer_isolation_level_mismatch_total — Gauge from a config-audit job; must be 0 in every environment before shipping a change to a downstream topic.
4. external_write_duplicate_detected_total — Counter on the dedup-store fallback path; a healthy non-zero rate confirms the fallback is actually catching redeliveries, not silently bypassed.
5. transaction_abort_due_to_rebalance_total — Counter; a rising rate means max.poll.interval.ms is too low relative to batch processing time.
6. Structured log fields on every transaction event — transactional_id, producer_epoch, partition, offset_range, event (begin | commit | abort) indexed for incident triage.

---

Related

• Exactly-Once vs At-Least-Once Delivery — parent page covering the precise definitions, the Two Generals Problem, and the implementation variants this runbook implements for Kafka specifically.
• Idempotency Fundamentals & API Guarantees — the parent section covering the full engineering contract for idempotent APIs and deduplication.
• Idempotent Consumer Patterns for Event Streams — the general dedup-store fallback pattern used in Step 4 above, independent of Kafka.
• Using Redis SETNX for Distributed Request Deduplication — a Redis-backed alternative to the Postgres dedup table used for external writes.
• Message Queue Deduplication Patterns — broker-agnostic deduplication strategies for teams running both RabbitMQ and Kafka.