Kafka Transactions vs Consumer-Side Deduplication

Kafka’s transactional producer offers a genuine exactly-once guarantee, and it is genuinely scoped: it covers records that live in Kafka. The moment a consumer’s work lands somewhere else — a database, a payment processor, an email gateway — the guarantee stops at the boundary and a deduplication table becomes mandatory again. This decision guide sits under message queue deduplication patterns.

Prerequisites. A Kafka cluster at 2.5 or later, and the framing from exactly-once vs effectively-once semantics.

Step 1 — Classify the consumer

Where the Kafka transaction boundary sits A shaded region labelled Kafka transaction encloses the input topic offsets, the consumer offsets topic, and the output topic. Outside the region, a PostgreSQL database and an external payment API are connected to the processing step by dashed arrows marked not covered. A caption states that anything crossing the boundary needs its own deduplication. Kafka transaction — atomic input topic payments.events process pure transform output topic ledger.entries __consumer_offsets committed in the same transaction payment processor API NOT rolled back on abort PostgreSQL ledger separate transaction Rule of thumb Effects entirely inside Kafka → transactions are sufficient and simpler. Any effect outside Kafka → you still need a deduplication table at the effect's store. Why transactional.id must be stable An instance is partitioned but still running. A replacement starts with the same transactional identifier, which bumps the producer epoch and fences the old instance so its writes are rejected. With a random identifier per process the old instance is never fenced, and both write to the output topic. old instance partitioned, still running replacement, same tx id epoch bumped → predecessor fenced ProducerFenced old writes rejected A random transactional.id per process fences nobody — both instances write.
Consumer shape Effects land in Correct mechanism
Stream transform (filter, join, aggregate) Kafka only Kafka transactions
Materialiser writing to PostgreSQL External database Deduplication table in that database
Notifier calling a payment API External service Idempotency key passed to the API
Hybrid: writes to Kafka and a database Both Transactions for Kafka, dedup table for the database

Step 2 — Configure read-process-write correctly

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "ledger-writer-" + instanceId);
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");  // critical
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);          // critical

producer.initTransactions();
while (running) {
    var records = consumer.poll(Duration.ofMillis(200));
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    try {
        for (var record : records) {
            producer.send(new ProducerRecord<>("ledger.entries", transform(record)));
        }
        // Offsets join the transaction — this is what makes it atomic end to end.
        producer.sendOffsetsToTransaction(offsetsOf(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        producer.close();          // a newer instance fenced us — do not retry
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();   // safe: nothing outside Kafka was touched
    }
}

Two settings do most of the work, and both are non-defaults. isolation.level=read_committed makes downstream consumers skip aborted records; without it the producer’s guarantee is discarded at the next hop. enable.auto.commit=false is what allows offsets to join the transaction; auto-commit would commit them independently and reintroduce the gap.

transactional.id must be stable per logical partition assignment, not random per process. A random id on restart means Kafka cannot fence the old instance, and zombie writes become possible.

Step 3 — Add a deduplication table for external effects

from psycopg.errors import UniqueViolation

def handle(record, conn) -> str:
    """The Kafka transaction cannot protect this write — the database must."""
    try:
        with conn.transaction():
            conn.execute(
                """INSERT INTO processed_events
                       (consumer_group, event_id, topic, partition, "offset")
                   VALUES (%s, %s, %s, %s, %s)""",
                (GROUP, record.key, record.topic, record.partition, record.offset),
            )
            apply_ledger_entry(record, conn)   # same transaction as the marker
        return "processed"
    except UniqueViolation:
        return "duplicate"

For an external API call, propagate a derived idempotency key so the remote system deduplicates:

# Deterministic: the same record always produces the same key, so a redelivery
# is collapsed by the processor rather than by us.
reversal_key = f"kafka:{record.topic}:{record.partition}:{record.offset}"
stripe.charges.create(amount=amount, source=source, idempotency_key=reversal_key)

Note the key is derived from topic:partition:offset rather than the record key, because a redelivery after a rebalance replays the same offset. If the topic can be recreated, prefix with a topic generation id — offsets restart at zero and would collide.

Step 4 — Measure the throughput cost

# Baseline: no transactions.
kafka-producer-perf-test --topic bench --num-records 2000000 --record-size 512 \
  --throughput -1 --producer-props acks=all bootstrap.servers=$BROKERS

# With transactions, committing every 1000 records.
kafka-producer-perf-test --topic bench --num-records 2000000 --record-size 512 \
  --throughput -1 --transactional-id bench-tx --transaction-duration-ms 100 \
  --producer-props acks=all bootstrap.servers=$BROKERS
Commit cadence Typical throughput cost Recovery window on crash
Every record 60–80% Near zero
Every 100 records 10–20% Up to 100 records reprocessed
Every 1,000 records 3–8% Up to 1,000 records reprocessed
Every 100 ms (time-based) 5–15% Up to 100 ms of records

The cost is dominated by commit frequency, not by the transactional machinery itself. Batching is the tuning knob, and it trades throughput against how much work is redone after a crash — which is harmless precisely because the transaction makes reprocessing safe.

Step 5 — Verify by killing the consumer

# Produce a known sequence, then kill mid-transaction and confirm no duplicates
# and no gaps in the output topic.
kafka-console-producer --topic payments.events --property parse.key=true \
  < seq_1_to_100000.tsv

sleep 5 && kill -9 "$(pgrep -f ledger-writer)"      # hard kill, mid-transaction
sleep 30                                            # let the group rebalance

kafka-console-consumer --topic ledger.entries --from-beginning \
  --isolation-level read_committed --timeout-ms 20000 \
  | awk -F'\t' '{print $1}' | sort -n | uniq -c \
  | awk '$1 != 1 {print "DUPLICATE:", $2}' | head
# expect: no output — every key appears exactly once
# And confirm the external effect was not duplicated, which the transaction
# does NOT protect. This is the check most teams skip.
psql -c "SELECT event_id, count(*) FROM ledger_entries
          GROUP BY event_id HAVING count(*) > 1 LIMIT 10;"
# expect: zero rows — because of the processed_events table, not because of Kafka

Failure scenarios and debugging

Failure Scenario Remediation Steps Observability Hooks
Downstream consumer left at isolation.level=read_uncommitted, so it reads records from aborted transactions Set read_committed on every consumer of a transactional topic and assert the setting at startup. It is not the default. kafka_consumer_isolation_level exported as a startup gauge; audit every consumer group
transactional.id generated randomly per process, so a restarted instance cannot fence its predecessor and zombie writes occur Derive the id deterministically from the group and the assigned partition range, and keep it stable across restarts. ProducerFencedException count — should be non-zero only during genuine handovers
Team assumes Kafka transactions cover a payment API call, and a rebalance charges a customer twice Pass a derived idempotency key to the external API and add a processed_events table for database effects. The transaction never covered it. Charges per distinct topic:partition:offset — must be exactly 1.0
Commit-per-record configuration cuts throughput by 70% and consumer lag grows without bound Batch 100–1,000 records per transaction, or commit on a 100 ms timer. Measure with kafka-producer-perf-test before and after. kafka_consumer_lag alongside transaction_commit_rate; lag growth with a high commit rate is the signature
Topic recreated, offsets restart at zero, and derived idempotency keys collide with keys from the previous generation Include a topic generation or creation timestamp in the derived key. Never assume offsets are globally unique over time. idempotency_key_collision_total at the external API; any value indicates key-space reuse
Extending the guarantee past the broker boundary An external call's idempotency key is composed from the topic generation, the topic name, the partition and the offset. The generation component is what stops a recreated topic, whose offsets restart at zero, from producing keys that collide with those of the previous generation. generation stops offset reuse : topic : partition : offset Deterministic: a redelivery of the same record produces the same key, so the processor collapses it.

SRE / observability checklist

  1. kafka_transaction_commit_rate and kafka_transaction_abort_rate — Counters per consumer group. An abort rate above 1% means the processing step is failing, not the transport.
  2. kafka_consumer_lag — Gauge per partition. Lag that grows only after enabling transactions points at commit cadence, not capacity.
  3. events_handled_total{outcome} — Counter split processed and duplicate from the deduplication table. A steady 0.01%–0.5% duplicate rate is normal redelivery.
  4. ProducerFencedException count — Should be near zero outside deployments. A steady rate means transactional.id is unstable.
  5. external_effect_per_offset_ratio — Derived: effects created divided by distinct topic:partition:offset. Must be exactly 1.0; this is the only metric that catches the boundary mistake.
  6. isolation.level and enable.auto.commit — Exported as startup gauges per consumer, so a misconfigured service is visible rather than assumed correct.