Kafka Exactly-Once Semantics: What Transactions Actually Guarantee

A staff engineer once told me his pipeline was exactly-once because he had set processing.guarantee=exactly_once_v2 in his Kafka Streams config. The topology consumed payments, enriched it, and wrote each record to Postgres with an INSERT. The Postgres table had 1.4 million rows against 1.39 million distinct payment IDs.

He was not wrong about the config. He was wrong about the boundary. Kafka’s exactly-once machinery is a closed system: it covers records read from Kafka, state held in Kafka-backed stores, offsets committed to Kafka, and records written back to Kafka. Postgres is outside that system, and the transaction coordinator has no idea that an INSERT happened.

That boundary is the whole subject of this post. I want to walk through what the idempotent producer actually fixes, what transactions add on top, why read_committed is not free, and what you do at the sink — because in every real pipeline I have built, the sink is where duplicates are actually prevented. If the partition and offset model is still new, the fundamentals post comes first.

Three delivery guarantees, and the one you usually have

At-most-once: commit the offset before processing. A crash loses records. Almost nobody wants this, but you get it accidentally with enable.auto.commit=true and a slow sink.

At-least-once: process, then commit. A crash between the two replays records. This is the default posture of every consumer that commits after work, and it is the correct default.

Exactly-once: each record affects the output exactly one time, regardless of retries and crashes. Kafka delivers this for Kafka-to-Kafka flows via three mechanisms working together: sequence numbers on the producer, an atomic transaction spanning produces and offset commits, and a consumer isolation level that hides uncommitted data.

Notice what the third guarantee requires: the output has to be something the transaction coordinator can roll back. Kafka topics qualify. Your warehouse does not.

The idempotent producer: turn it on and forget it

Before transactions, there is a duplicate source that has nothing to do with your application logic. A producer sends a batch, the broker writes it and replicates it, and the acknowledgement is lost to a network blip. The producer retries. The broker writes the batch again. One send, two records, and no code you wrote is at fault.

The idempotent producer fixes this at the protocol level. Each producer gets a producer ID, each partition gets a monotonic sequence number, and the broker deduplicates any batch whose sequence it has already seen.

Properties p = new Properties();
p.put("bootstrap.servers", "b-1.msk.internal:9092");
// Default true in clients 3.0+. Set it explicitly so a config
// audit shows intent rather than a version-dependent default.
p.put("enable.idempotence", "true");
// Implied by idempotence, but be explicit: acks=all means the
// leader waits for all in-sync replicas before acknowledging.
p.put("acks", "all");
p.put("max.in.flight.requests.per.connection", "5"); // ordering is preserved up to 5
p.put("retries", Integer.MAX_VALUE);
p.put("delivery.timeout.ms", "120000"); // this, not retries, is the real bound

Two things worth internalizing. First, max.in.flight up to five is safe only with idempotence enabled; without it, a retried batch can land after a later batch and silently reorder your partition. Second, retries is not the knob that bounds how long a send can take — delivery.timeout.ms is. Setting retries to a huge number with a sane delivery timeout is the correct combination.

Also understand the scope: idempotence is per producer session, per partition. If the process restarts and gets a new producer ID, the broker cannot deduplicate against the previous session. That gap is exactly what transactions with a stable transactional.id close.

Transactions: atomic produce plus offset commit

A transactional producer wraps a set of produces and the offset commit for the records that caused them into one atomic unit. Both happen or neither does.

The pattern is consume-transform-produce, and the critical detail is that the offset commit goes through the producer, not the consumer:

producer.initTransactions(); // fences any zombie with the same transactional.id

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

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, Order> rec : records) {
            Enriched out = enrich(rec.value());
            producer.send(new ProducerRecord<>("orders.enriched", rec.key(), out));
        }
        // The offsets are part of the transaction. This is the whole trick.
        producer.sendOffsetsToTransaction(
            offsetsOf(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        // Another instance took over our transactional.id. We are the zombie.
        producer.close();
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction(); // consumer will re-read these offsets
    }
}

transactional.id must be stable across restarts and unique per logical producer instance. On restart, initTransactions() contacts the transaction coordinator, bumps the producer epoch, and fences the old incarnation — any in-flight writes from the previous process are rejected. That is how a hung-then-recovered pod cannot corrupt the output.

The consumer in this loop must have enable.auto.commit=false. If auto-commit is on, offsets move outside the transaction and the atomicity is a fiction.

sequenceDiagram
  participant C as Consumer
  participant P as "Transactional producer"
  participant TC as "Transaction coordinator"
  participant T as "Output topic"
  C->>P: "Batch of 200 records"
  P->>TC: beginTransaction
  P->>T: "Write enriched records"
  P->>TC: "sendOffsetsToTransaction"
  P->>TC: commitTransaction
  TC->>T: "Write commit marker"
  Note over T: "read_committed consumers see records now"

Offsets ride inside the transaction. Nothing is visible downstream until the commit marker lands.

read_committed is not free

Producing transactionally is half the contract. The downstream consumer must set isolation.level=read_committed, or it reads aborted records as if they were real and the whole exercise was theater.

read_committed consumers stop at the last stable offset — the highest offset before the earliest still-open transaction. This has a latency consequence that surprises people: a single long-running transaction on a partition blocks visibility of every record after it, including records from transactions that already committed.

If your transaction commits every 100 ms, added latency is around 100 ms. If one consumer instance batches for 30 seconds before committing, everything behind it waits 30 seconds. I size transactions to commit at least every second, and I alert on transaction.abort.rate because aborts leave markers that consumers must skip.

transaction.timeout.ms (default 60 seconds, capped by the broker’s transaction.max.timeout.ms) is the coordinator’s patience. Exceed it and the transaction is aborted for you, which you discover as a ProducerFencedException on the next send.

The sink is where duplicates actually get prevented

Now the part that matters for data engineering work. The moment the output goes anywhere other than a Kafka topic — Postgres, Snowflake, S3, Elasticsearch, an HTTP endpoint — the transaction coordinator is out of the picture. There is no two-phase commit between Kafka and your warehouse, and pretending otherwise is how you get 1.4 million rows for 1.39 million payments.

What works instead is making the write idempotent on a key the producer already guarantees. Then at-least-once delivery plus an idempotent sink equals effectively-once output, which is the property you actually wanted.

The natural key is best when the source has one:

-- Snowflake sink. order_id is the business key; the stream may
-- deliver the same record twice after a rebalance or a restart.
MERGE INTO analytics.raw.orders AS t
USING (
    SELECT
        order_id,
        customer_id,
        status,
        amount,
        updated_at,
        -- Keep only the newest version per key inside this batch.
        ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY updated_at DESC) AS rn
    FROM @kafka_stage/orders/batch_2026_05_07_1412.parquet
    QUALIFY rn = 1
) AS s
ON t.order_id = s.order_id
-- Late-arriving replays must not overwrite newer state.
WHEN MATCHED AND s.updated_at > t.updated_at THEN UPDATE SET
    t.customer_id = s.customer_id,
    t.status      = s.status,
    t.amount      = s.amount,
    t.updated_at  = s.updated_at
WHEN NOT MATCHED THEN INSERT
    (order_id, customer_id, status, amount, updated_at)
VALUES
    (s.order_id, s.customer_id, s.status, s.amount, s.updated_at);

Two guards in there earn their keep. The QUALIFY rn = 1 deduplicates within the batch, because a replayed micro-batch frequently contains the same key twice. The s.updated_at > t.updated_at predicate makes the MERGE order-independent — a replay of an older record leaves the table alone instead of resurrecting stale state. That second condition is the one teams forget, and it turns a correct-looking MERGE into a time machine.

When the source has no natural key, synthesize a deterministic one from coordinates that never change for a given record:

import hashlib

def event_id(topic: str, partition: int, offset: int) -> str:
    """Deterministic per-record identity.

    Topic, partition, and offset uniquely identify a record for the
    life of the log, so a replay of the same record produces the same
    id and collides with the row already written.
    """
    raw = f"{topic}:{partition}:{offset}".encode()
    return hashlib.sha256(raw).hexdigest()[:32]


def write_batch(cur, topic: str, rows: list[dict]) -> int:
    """Insert with an idempotency key; duplicates become no-ops."""
    payload = [
        (event_id(topic, r["partition"], r["offset"]), r["key"], r["value_json"])
        for r in rows
    ]
    cur.executemany(
        """
        INSERT INTO events_raw (event_id, event_key, payload)
        VALUES (%s, %s, %s)
        ON CONFLICT (event_id) DO NOTHING
        """,
        payload,
    )
    return cur.rowcount

The offset-based key has one caveat: it is stable only if you never republish the same logical event to a new offset. For a compacted topic or a topic you replay into from a backfill job, use the business key instead. Whichever you pick, write it down in the data contract so the sink team and the producer team agree on what “the same record” means.

flowchart TD
  src["Source topic"] --> app["Consume + transform"]
  app --> out["Output topic"]
  out --> boundary{"Inside Kafka?"}
  boundary -->|yes| eos["Transactions + read_committed"]
  boundary -->|no| sink["External sink"]
  sink --> idem["MERGE or ON CONFLICT on an idempotency key"]
  idem --> table["Warehouse table"]

The dashed line in your head should be at the Kafka boundary. Transactions stop there; the sink takes over.

What exactly-once costs

Nothing here is free, and the cost is worth naming before you turn it on everywhere.

Idempotent producer: effectively zero. A few bytes of sequence metadata per batch. Turn it on universally.

Transactions: extra round trips to the transaction coordinator per transaction, commit markers written into every partition you touched, and a throughput hit that in my measurements runs somewhere between 3% and 20% depending on how small your transactions are. Small transactions cost more per record; large ones cost latency downstream through read_committed.

read_committed consumers: added end-to-end latency equal to your transaction duration, plus buffering of aborted records that get discarded client-side.

Operational surface: transactional.id becomes part of your deployment identity. Two pods sharing one ID will fence each other in a loop. One pod per ID, derived from a stable ordinal, same discipline as group.instance.id in the rebalancing post.

Pitfalls

Believing exactly_once_v2 covers external writes. It covers Kafka topics and Kafka-backed state stores. A JDBC call inside a process() method is outside the transaction and will be executed again on replay.

Producing transactionally while consumers read uncommitted. The default isolation.level is read_uncommitted. If you do not change it downstream, aborted records are delivered and you have paid the transaction cost for nothing.

Auto-commit left on in a transactional consumer. Offsets then move independently of the transaction, and a crash replays or skips depending on timing. Set enable.auto.commit=false.

Reusing one transactional.id across instances. Each initTransactions() bumps the epoch and fences the other one. Two pods with the same ID produce a fencing ping-pong that looks like a broker fault.

A MERGE without a recency predicate. Idempotent against duplicates, but an out-of-order replay overwrites new state with old. Always compare updated_at or a monotonic version.

Very long transactions to “reduce overhead”. You move the cost from producer throughput to consumer latency, and one stuck transaction blocks the last stable offset for every downstream reader on that partition.

FAQ

Do I need transactions if my sink is already idempotent?

Usually not. An idempotent producer plus at-least-once consumption plus an idempotent sink gives you the same observable outcome with less operational surface. I reserve transactions for topologies that read from Kafka and write back to Kafka, especially multi-topic fan-out where partial writes would be visibly inconsistent.

Does exactly-once work across two Kafka clusters?

No. A transaction is scoped to one cluster’s transaction coordinator. MirrorMaker 2 and most replication tools are at-least-once across clusters, so a cross-cluster pipeline needs sink-side deduplication regardless of what each cluster does internally.

How does this compare to Kinesis?

Kinesis has no transaction concept at all. You get at-least-once delivery and you build idempotency into the consumer, which is also how I would build a Kafka pipeline with an external sink. The broader trade-offs are in Kafka vs Kinesis.

What about Kafka Connect sinks?

Connect sink tasks are at-least-once by default. Some connectors support exactly-once through upsert semantics on a configured key, which is the MERGE pattern above wearing a config file. Check whether your connector’s insert.mode is insert or upsert before assuming anything; the details are in Kafka Connect in production.

Can I get exactly-once with a file sink to S3?

Only by making the write atomic and named deterministically. Write to a temporary key, then copy to a final key derived from the topic, partition, and start offset. A replay produces the same final key and overwrites itself harmlessly. Appending to an existing object does not work.

Is the throughput cost worth it?

If both ends are Kafka and correctness matters more than 10% of throughput, yes. If either end is external, you are paying the cost and still need sink idempotency, so spend the effort on the sink instead.

What this means for your pipelines

Enable the idempotent producer everywhere today. It is on by default in current clients, it eliminates a real duplicate source that has nothing to do with your code, and it costs nothing. If any of your producers are on an older client or explicitly set enable.idempotence=false, that is a one-line fix worth shipping this week.

Then draw the Kafka boundary on your architecture diagram and be honest about where each pipeline crosses it. Kafka-to-Kafka topologies — stream joins, enrichment chains, fan-out to several downstream topics — are where transactions and read_committed earn their overhead, and where Kafka Streams gives you the whole thing behind one config value. Everything that terminates in a warehouse, a search index, a cache, or an HTTP API is at-least-once in practice no matter what the producer config says.

For those pipelines, put your effort into the idempotency key. Pick it deliberately, prefer a business key over a coordinate key, add the recency predicate so replays cannot rewrite newer state, and document it where both the producer and the consumer team can see it. A sink that is idempotent by construction survives rebalances, restarts, backfills, and the inevitable day someone replays a week of history to fix an enrichment bug. That resilience is worth more than any config flag.