Exactly-Once Semantics

Architecture

A single transaction touches three brokers (or roles): the producer, the transaction coordinator, and the data partition leaders.

Key Numbers

Delivery Semantics: At-Most-Once, At-Least-Once, Exactly-Once

Three delivery semantics sit on a spectrum between "don't lose data" and "don't duplicate data." Kafka defaults to at-least-once with acks=all, and exactly-once is opt-in via the transactions API.

At-Most-Once

At-most-once means a record is never delivered twice — but it might be lost. The sender fires and forgets: acks=0 or acks=1 with aggressive retries disabled. If the broker goes down before acknowledgment, the record is gone. Downstream systems never see duplicates, but they will miss events. This is useful for low-value telemetry where losing a reading is acceptable, but catastrophic for financial transactions or inventory updates.

At-Least-Once

At-least-once is Kafka's default with acks=all and retries enabled. Every record is guaranteed to be written to disk at least once. If a produce request fails, the producer retries. The broker may have already written the first attempt — retries are indistinguishable from the original on the broker side. The result: duplicates are possible. This is the correct choice when downstream systems can safely deduplicate by record key. Most ETL pipelines, log aggregation, and metrics collection use at-least-once.

The critical implication: at-least-once requires idempotent downstream consumers. Your pipeline must be safe to process the same record twice. If you're writing to a relational database, you need an upsert (INSERT ... ON CONFLICT DO NOTHING) or a deduplication key. If you're incrementing a counter, you need exactly-once semantics upstream.

Exactly-Once

Exactly-once in Kafka means: each input record contributes to the output state exactly once, even if the pipeline crashes, restarts, or retries. This is the strongest guarantee and the most expensive. It requires all three of:

• Idempotent producer — retries don't produce duplicates on the broker side
• Transactions — writes across partitions plus offset commits are atomic
• read_committed consumers — aborted records are filtered out

Without transactions, an at-least-once producer can still cause double-delivery in the read-process-write loop: a crash after producing to the output topic but before committing the consumer offset causes the offset to be rewound on restart, re-emitting all records since the last committed offset.

Why Exactly-Once Is Hard

Distributed exactly-once is theoretically impossible without a distributed transaction protocol. The FLP impossibility result (Fischer, Lynch, and Paterson, 1985) proves that no deterministic consensus algorithm can guarantee agreement in an asynchronous network with even one possible failure. Kafka's exactly-once is a form of "exactly-once semantics for the application layer" — not a guarantee that a record crosses the wire once, but that its effect on state is applied once.

The Three Failure Modes That Break At-Least-Once

In a distributed system, failures are not rare events you plan around — they are the steady state. Exactly-once must survive three categories of failure simultaneously:

• Producer-side retried sends: A produce request times out at the TCP level or returns a NotEnoughReplicasException. The producer retries with the same sequence number. Without idempotence, the broker writes a duplicate. With idempotence, the broker detects and deduplicates, but this only covers one producer instance. If the producer JVM restarts, it gets a new PID and the idempotency window doesn't carry over.
• Transactional coordinator failure: The broker running the Transaction Coordinator crashes after the coordinator wrote data to partitions but before it wrote CompleteCommit to __transaction_state. When that broker recovers, the log replay must recover the correct state. This is why PrepareCommit is the decision point — once durable, the transaction cannot be lost.
• Zombie producers: A producer instance hangs in a long GC pause, is considered dead by the orchestrator, and a new instance takes over with the same logical identity. The old instance wakes up and tries to commit or produce. Without fencing, two producers with the same transactional.id would race, producing divergent output. With fencing, the old producer's epoch is superseded and its produce requests are rejected.

The CAP Theorem Context

Kafka with acks=all is a CP system (Consistent + Partition-tolerant) — it will not acknowledge writes until all in-sync replicas confirm, and it will block if a quorum is lost. With transactional.id fencing, Kafka guarantees consistency: at most one producer with a given identity can make progress. The availability tradeoff is that if the Transaction Coordinator broker is unavailable, no producer with that transactional.id can start a transaction — which is the correct behavior for a system that must not produce duplicate records.

Kafka Streams adds another layer: the exactly_once_v2 processing guarantee requires that the consumer group coordinator (a Kafka broker) and the Transaction Coordinator are the same broker for the partition that owns the consumer group offsets. This colocation is what makes atomic offset commits possible. When a Streams task fails over to another instance, the new instance's consumer rewinds to the last committed offset — which is the same offset the producer committed transactionally — and reprocesses. Because the output writes are idempotent within the transaction, reprocessing produces the same result.

The Problem with At-Least-Once in Practice

Consider a payment processor: it consumes a payments topic, debits account A, credits account B, and produces a receipts topic. With at-least-once, if the producer fails after writing the receipt but before committing the consumer offset, on restart the payment is re-processed — account A is debited twice. The consumer deduping strategy (filter by payment ID) helps at the consumer level, but:

• The debit/credit has already happened against the external database
• Deduplicating at read time doesn't undo the first debit
• The external system (bank ledger) is not transactional with Kafka

Exactly-once for external systems requires transactional outbox pattern or Change Data Capture with idempotent applied at the sink. Kafka's EOS only covers Kafka-to-Kafka pipelines. We'll cover this limitation in detail in the EOS with External Systems section.

The Idempotent Producer: PID + Epoch + Sequence

The idempotent producer is Kafka's first line of defense against duplicates. It is a per-producer-instance, per-partition deduplication mechanism that requires no coordination with the Transaction Coordinator. It works entirely in the broker's producer state cache.

How It Works

When a producer with enable.idempotence=true starts, it calls initProducerId() (internally, triggered by the first send or explicitly via initTransactions()). The broker assigns a producer ID (PID), a starting epoch of 0, and a transactional ID mapping if one was provided. This PID is unique per producer JVM lifetime — a new JVM restart gets a new PID.

Every produce request then carries: (PID, epoch, partition, sequenceNumber). The sequence number is a 32-bit signed integer that monotonically increases per-partition. The broker (specifically, the partition leader) maintains a map of (PID, partition) → (lastAcceptedSequence, lastEpoch).

Deduplication Rules on the Broker

When a batch arrives at the leader, three cases are possible:

• seq == lastSeq + 1: Normal case. The batch is accepted, written, and lastSeq is updated to seq. The broker returns the base offset to the producer.
• seq == lastSeq: Duplicate. The broker returns the same base offset as the original write without re-writing. This is the idempotent dedup in action.
• seq < lastSeq + 1: Out-of-order sequence. The broker rejects with OutOfOrderSequenceException. This happens when a batch that was queued for retry is returned out of order (e.g., a later batch's ack arrived first). The producer must call producer.abortableTransaction() to reset state and start a new epoch.
• epoch < lastEpoch: Stale epoch (older producer trying to write). The broker rejects with ProducerFencedException. This is the fencing path.

The Idempotency Window

The broker tracks the last 5 sequence numbers per (PID, partition). This is controlled by max.in.flight.requests.per.connection=5 — when you set idempotence=true, Kafka forces this to 5, which is the maximum that still allows the broker to distinguish 5 concurrent in-flight batches per partition. The producer state cache has a maximum size of max.produce.requests.catches and evicts least-recently-used PIDs when full.

The implication: if you have more than 5 in-flight batches for the same partition from the same producer, the broker may evict the oldest sequence from its cache, and a late retry could be accepted as a new record rather than deduplicated. Setting max.in.flight.requests.per.connection above 5 with idempotence enabled will throw a config exception.

Producer-initiated PID Assignment

The PID is assigned by the broker on InitProducerId and stored in the producer's local state. The producer also stores the epoch and per-partition sequence numbers locally in an ApiResult<InitProducerIdResult>. This means:

• If the producer hard-kills and restarts, it calls InitProducerId again and gets a new PID. The old PID's sequence is orphaned in broker state, but the new PID has a fresh sequence start.
• If the producer crashes but the broker still believes the PID is alive, the producer can still use the same PID on reconnect if transactional.id is not set. With transactional.id, the reconnect triggers epoch bump and fencing.

// Producer config — idempotent
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092");
props.put("enable.idempotence", "true");
// acks=all and max.in.flight=5 are forced internally when idempotence=true
// but being explicit documents intent:
props.put("acks", "all");
props.put("max.in.flight.requests.per.connection", "5");
props.put("retries", Integer.MAX_VALUE);
// retry backoff to avoid thundering herd on transient errors
props.put("retry.backoff.ms", 100);
props.put("request.timeout.ms", 30000);

KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord<String,String> record = new ProducerRecord<>("orders", key, value);

// The metadata returned here carries the offset and partition
Future<RecordMetadata> future = producer.send(record);
RecordMetadata metadata = future.get(); // blocks
long offset = metadata.offset();

The max.in.flight=5 Constraint

With a maximum of 5 in-flight requests per connection, and a per-partition monotonic sequence, the broker can deduplicate a window of 5 batches. This is a tradeoff: larger windows mean more memory in the broker's producer state cache. The number 5 was chosen empirically as a sweet spot that covers most batching patterns while keeping memory bounded.

The consequence for throughput: a single producer thread can have at most 5 batches in flight per partition. With a typical batch size of 16KB, this means the producer can have 80KB of unacknowledged data per partition. For a high-throughput pipeline writing to many partitions in parallel, this is not a bottleneck since each partition is independent and the 5-limit applies per-partition, not per-producer.

Transactional Producers: transactional.id, acks, and Isolation

Idempotence alone covers a single producer instance and a single partition's writes. Transactions cover multi-partition atomicity and producer restarts. Together they form the complete exactly-once story.

The transactional.id Configuration

The transactional.id is a user-chosen string that identifies a logical producer instance across JVM restarts. It must be stable per logical processor — that is, if you have 4 instances of your processor running, each should have a distinct transactional.id (e.g., "order-processor-0" through "order-processor-3", or derived from a Kubernetes pod ordinal).

The coordinator uses the transactional.id to pick the __transaction_state partition via Utils.abs(transactional.id.hashCode()) % numPartitions. This colocation means all transaction state for a given logical producer lives on the same broker — no distributed locking needed.

When initTransactions() is called, the coordinator looks up the current epoch for this transactional.id and bumps it. Any produce request with the old epoch from a zombie producer is rejected with ProducerFencedException.

transactional.id vs. PID

The PID and the transactional.id serve different purposes:

• PID is assigned by the broker on InitProducerId. It is opaque and unique per producer JVM lifetime. Used by the idempotent producer dedup logic.
• transactional.id is chosen by the application developer. It is stable across JVM restarts for the same logical instance. Used by the Transaction Coordinator for fencing and for mapping to the coordinator partition.

A producer can have a PID without a transactional.id (using only idempotence). It cannot have a transactional.id without also having idempotence enabled (Kafka enforces this).

transaction.timeout.ms

This is the maximum time a transaction can be open before the coordinator aborts it automatically. Default is 15 seconds. If commitTransaction() is not called within this window, the coordinator writes PREPARE_ABORT to __transaction_state, issues abort markers to all involved partitions, and completes the abort.

This timeout protects against producers that hang mid-transaction — for example, a producer that is slow at processing and never calls commitTransaction(), or one that crashes while a transaction is open. The coordinator cleans up stale transactions via a background task that scans for expired transactions.

The tradeoff: if your processing per batch takes longer than transaction.timeout.ms, the coordinator will abort while you're still working. Set this generously for slow processors, but be aware that a long timeout means longer exposure to the LSO stalling issue (discussed below).

Interaction with max.poll.interval.ms

max.poll.interval.ms is a consumer config that controls the maximum time between poll() calls. If the consumer doesn't call poll() within this window, the consumer group coordinator triggers a rebalance, and the partition is handed to another instance. This interacts with EOS in two ways:

• The producer's open transaction stalls the consumer's poll cycle. If a producer calls beginTransaction() and then takes longer than max.poll.interval.ms to call commitTransaction(), the consumer group thinks this instance is dead and revokes its partitions. The new instance takes over and starts consuming from the last committed offset — which may be before the records that the old instance was processing. If the old instance had produced some records in the open transaction, those records are now in an orphaned transaction that the coordinator will eventually abort.
• The fix: Keep the transaction short (under max.poll.interval.ms), or use max.poll.interval.ms large enough to accommodate your worst-case processing time plus the transaction overhead. As of Kafka 3.x, the consumer has a gapless_divide_partitions option to prevent rebalance during a transaction, but this is still experimental.

Exactly-Once With Foreign Systems

Kafka transactions do not span external systems. If your pipeline produces to a Kafka topic and writes to a database in the same logical operation, a crash between the two writes will result in one side succeeding and the other not. This is a fundamental limitation of the 2-phase commit problem applied to heterogeneous systems.

The canonical solution is the Transactional Outbox Pattern:

1. Write the record AND the outbox entry in the same local database transaction
2. A separate relay process reads the outbox table and publishes to Kafka
3. Because the database commit is the commit point, the outbox entry is idempotent (relay can retry)
4. On the consumer side, write the consumed record + offset to the database in the same transaction

This is how Kafka EOS is extended to external systems: not by making Kafka transactional with the DB, but by making the DB the commit point for the local transaction and using Kafka idempotence for the wire.

Consumer Transactions: The Read-Process-Write Pattern

The read-process-write pattern is the heart of exactly-once stream processing. It couples the consumer offset commit with the producer output write in a single atomic transaction.

The Fundamental Problem

Without transactions, the read-process-write loop has an inherent gap: reading and processing happens at one point in time, and offset committing happens later (via the consumer's commitSync() or auto-commit). If the producer writes to the output topic but the consumer crashes before committing the offset, on restart the consumer rewinds to the last committed offset — which is before the records just processed — causing double-delivery.

The reverse is also bad: if you commit the offset before processing, and the producer crashes after committing but before the output is durable, you lose a record and the offset no longer matches the visible state.

Atomic Offset Commit via sendOffsetsToTransaction

producer.sendOffsetsToTransaction() is the bridge. It writes the consumer offsets to the __consumer_offsets topic as a transactional record. The offsets topic partition used is derived from the consumer's groupMetadata (specifically the groupInstanceId if provided, otherwise the memberId). This write happens as part of the same transaction as the output records — so either all succeed or all abort.

When the transaction commits, two things become visible atomically:

• The output records written to the output topic
• The offset commit in __consumer_offsets

A consumer with isolation.level=read_committed sees the new output records and the new offset at the same moment. A consumer with isolation.level=read_uncommitted sees output records as soon as they are written to the leader (before the transaction commits), but may also see aborted records.

Consumer Group Metadata and Partition Ownership

The sendOffsetsToTransaction call takes a ConsumerGroupMetadata object that encodes the consumer's group ID, generation ID, member ID, and instance ID (if using static membership). This metadata is written into the offset commit record and used by the consumer group coordinator to validate that the offset commit is coming from the legitimate owner of the partition.

If a consumer instance with static membership leaves the group and rejoins, it presents the same groupInstanceId and gets the same partitions assigned. The generation ID increments on each join, which prevents stale offset commits from a prior generation from being accepted.

Idempotent Consumers

Even with EOS, consumers should be idempotent when possible. The reason: the read_committed consumer sees committed records. If a transaction aborts after some of its records were visible to a read_uncommitted consumer (which happens when a downstream consumer has isolation.level=read_uncommitted), those records must be invisible to read_committed consumers. The log segment stores an abort index that maps transaction IDs to their outcome (commit or abort). The consumer fetcher uses this to filter out aborted records.

For consumers that cannot easily deduplicate (e.g., incrementing a counter), consider using the record key as a deduplication key in an upsert database operation. The exact-once guarantee from Kafka protects the transport; the downstream system's upsert is the last line of defense.

// Full read-process-write loop with exactly-once
props.put("bootstrap.servers", "kafka-1:9092,kafka-2:9092");
props.put("group.id", "order-enricher-group");
props.put("isolation.level", "read_committed");

props.put("enable.auto.commit", "false"); // we commit offsets in the transaction
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
KafkaProducer<String,String> producer = new KafkaProducer<>(producerProps);

producer.initTransactions();

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

  producer.beginTransaction();
  try {
    Map<TopicPartition,OffsetAndMetadata> offsets = new HashMap<>();
    for (ConsumerRecord<String,String> r : records) {
      // Enrich: join with reference data, transform
      String enriched = enrich(r.value());
      // Produce to output topic transactionally
      producer.send(new ProducerRecord<>("orders-enriched", r.key(), enriched));
      // Track offset to commit: next offset after this record
      offsets.put(new TopicPartition(r.topic(), r.partition()),
                  new OffsetAndMetadata(r.offset() + 1, r.partition()));
    }
    // Atomic: write output + commit offsets together
    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
    producer.commitTransaction();
  } catch (ProducerFencedException e) {
    // Another producer took our transactional.id — we're fenced out
    producer.close();
    throw new IllegalStateException("Fenced out of transaction group", e);
  } catch (KafkaException e) {
    // Non-fatal error (e.g., broker went down mid-txn): abort and retry
    producer.abortTransaction();
    // Optional: backoff before retry
    Thread.sleep(1000);
  }
}

EOS with Kafka Streams: exactly_once and exactly_once_v2

Kafka Streams is the most common consumer of EOS. It uses the transactions API internally to provide exactly-once guarantees across state stores, output topics, and consumer offsets.

Processing Guarantees

Kafka Streams supports three processing guarantees, set via processing.guarantee=exactly_once_v2 (default since Kafka 2.5):

• at_least_once (default before 2.5): Commits offsets after processing, may reprocess on restart, but output is written at least once. State stores are updated via changelog compaction.
• exactly_once: Uses the transactions API. Offsets and output writes are in the same transaction. State store updates are also written transactionally to their changelog topics. This prevents duplicate state store records but adds overhead.
• exactly_once_v2 (current default): Improved implementation that avoids unnecessary transaction aborts and reduces coordinator overhead. Uses the same underlying protocol as exactly_once but with better handling of consecutive tasks and reduced produce pressure.

How exactly_once_v2 Works Internally

In exactly_once_v2, each task has a dedicated transactional producer that is initialized with a unique transactional.id per task. The StreamsTask manages the transaction lifecycle:

1. On poll(), the task records its read position (the consumer position)
2. Records are processed, updating state stores
3. Before producing, the task calls beginTransaction()
4. All output records (to output topics and state store changelogs) are sent in the transaction
5. commitTransaction() atomically commits offsets + output + changelog writes

The key insight: state stores in Kafka Streams are backed by changelog topics (RocksDB snapshots written to Kafka). When exactly-once is enabled, these changelog writes are also transactional. This means that if a task fails and another instance takes over, it can rebuild the state store by consuming from the changelog topic — and the changelog consumption is also within a transaction, so the rebuild is consistent with the offset position.

State Store Changelog and Recovery

Every Kafka Streams state store (e.g., StoreBuilder with withLoggingEnabled=true) has a corresponding changelog topic. When state is updated, the change is appended to the changelog topic as a KeyValue record. With EOS, these changelog writes are transactional with the output and offset commits.

Recovery from failure: when a task fails over, the new instance reads the compacted changelog topic from offset 0 (or from the last committed offset if usingEOS) and rebuilds the state store. Because the changelog is consumed transactionally, the rebuilt state is consistent with the consumer offset position. Without EOS, the changelog writes are non-transactional and a crash between the state update and the offset commit could cause the new instance to re-apply state updates that were already flushed.

Topology Test Driver

The TopologyTestDriver is an in-process test driver for Kafka Streams topologies. It simulates the Streams pipeline without requiring a Kafka cluster. With EOS enabled, the test driver uses an in-memory transactional producer that mirrors the same transaction lifecycle as the real Streams producer.

The test driver supports:

• Piped input and output (via TopologyTestDriver.pipeInput())
• State inspection after each pipe operation
• Transaction lifecycle hooks to verify commit/abort behavior
• Windowed state store verification (session windows, tumbling windows)

// Test with TopologyTestDriver and EOS
StreamsBuilder builder = new StreamsBuilder();
KStream<String,Order> orders = builder.stream("orders",
    Consumed.with(Serdes.String(), new OrderSerde()));

orders.filter((k, v) -> v.getAmount() > 100)
      .mapValues(order -> enrich(order))
      .to("high-value-orders", Produced.with(Serdes.String(), new OrderSerde()));

Topology topology = builder.build();
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor-test");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,
          Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
          new OrderSerde().getClass());

TopologyTestDriver driver = new TopologyTestDriver(topology, props);

TestInputTopic<String,Order> input = driver.createInputTopic("orders",
    new StringSerializer(), new OrderSerializer());
TestOutputTopic<String,Order> output = driver.createOutputTopic("high-value-orders",
    new StringDeserializer(), new OrderDeserializer());

input.pipeInput("order-1", new Order("order-1", "CUSTOMER_A", 250.00));
Order result = output.readValue();

assertEquals("CUSTOMER_A", result.getCustomerId());
assertEquals(250.00, result.getAmount(), 0.01);

// Verify state store
ReadOnlyKeyValueStore<String,CustomerProfile> store =
    driver.getKeyValueStore("customer-profiles");
assertNotNull(store.get("CUSTOMER_A"));

driver.close();

Consumer Group Coordination with EOS

When a Kafka Streams instance with EOS crashes and another instance takes over its tasks, the new instance must rebuild state from the changelog and resume from the last committed offset. This is coordinated by the consumer group protocol: the new instance joins the consumer group, is assigned the same partitions, and begins consuming from the last committed offset (stored in __consumer_offsets by the prior instance's transaction).

The consumer group coordinator (a Kafka broker) handles the rebalance. With static membership (group.instance.id), the new instance presents the same instance ID and gets the same partition assignment without triggering a rebalance — reducing downtime. With dynamic membership, a full rebalance occurs, which involves revocation and re-assignment of all partitions.

exactly_once_v2 vs. exactly_once: What's Different

The primary difference is how consecutive transactions are handled. In exactly_once, each transaction is followed by a commitTransaction() and then a new beginTransaction(). In exactly_once_v2, the producer keeps the transaction open across multiple poll cycles when possible, batching more records per transaction and reducing the per-record overhead of the two-phase commit. This reduces the number of transactions per second and the coordinator load.

exactly_once_v2 also handles the case where a task's processing is so fast that the transaction overhead dominates. It uses a "pipelined" approach where produce requests are issued without blocking on each individual batch's acknowledgment, reducing end-to-end latency while maintaining exactly-once guarantees.

EOS with External Systems: Sink Connectors and Delivery Guarantees

Kafka transactions cover Kafka-to-Kafka pipelines only. When the sink is a database, object store, or search index, a different approach is needed. The Kafka Connect framework provides exactly-once delivery for connectors through the IdempotentProducer and transactional configuration.

The Fundamental Limitation

Kafka's transaction protocol cannot span heterogeneous systems. The coordinator can write commit/abort markers to Kafka partitions, but it cannot tell an Oracle database to commit or abort. Two-phase commit across heterogeneous systems requires a distributed transaction coordinator (XA), which Kafka does not support and which most modern systems avoid due to its latency and availability cost.

The practical consequence: a sink connector that writes to a database and then crashes, with the Kafka transaction committed but the DB write not yet durable, will cause data inconsistency. The reverse — DB write succeeds, Kafka transaction aborts — is also possible if the connector crashes after the DB write but before the offset commit.

Making Sinks Idempotent

The standard approach is to make the sink idempotent, accepting that the Kafka-side read-process-write guarantees exactly-once delivery to the connector, and the connector handles idempotent application to the sink.

For most databases, this means using upsert semantics — INSERT ... ON CONFLICT DO UPDATE in PostgreSQL, MERGE in Snowflake, UPSERT in BigQuery. The record key is used as the deduplication key so that if the connector receives the same record twice, the second application is a no-op (or updates to the same values).

Change Data Capture (CDC) with Idempotent Apply

Debezium and similar CDC tools capture changes from databases as Kafka records. With EOS enabled, each change record is delivered exactly once to the sink connector. The connector applies changes using the source's transaction ID or monotonic timestamp as an idempotency key. MySQL binlog positions, PostgreSQL LSNs, and Oracle SCNs are all monotonic identifiers that let the sink skip already-applied changes.

S3 Sink Considerations

Writing to S3 via Kafka Connect S3 connector requires special handling for exactly-once. S3 is append-only object storage — there is no native upsert. The strategies are:

• File-level commit with deduplication: The connector batches records into files and writes each file with a unique name derived from the Kafka topic, partition, and start offset. If the connector restarts mid-upload, the partial file is discarded and re-uploaded. Files are named with offsets (e.g., topic-partition-startOffset-endOffset-timestamp.json), so re-uploading the same file to S3 is safe (S3 is eventually consistent, so overwriting with the same content is fine).
• Exactly-once delivery to S3 via transaction markers: The connector uses a separate producer for S3 writes and coordinates offset commits separately. If the S3 write fails, the offset is not committed, so records are redelivered. The file naming strategy prevents double-writing at the S3 object level.

Two-Phase Commit: The Coordinator and the Offset Topic

Kafka's transaction protocol is a variant of two-phase commit (2PC) adapted for Kafka's log-based architecture. The __transaction_state topic serves as the commit log, and the WriteTxnMarker protocol writes commit/abort markers to data partitions.

Why Two-Phase Commit

A naive atomic commit would write to all partitions and __consumer_offsets in one RPC — but that requires all partition leaders to be available simultaneously and would block on network partitions. Two-phase commit separates the decision (phase 1) from the action (phase 2):

1. Phase 1 — Prepare: The coordinator asks all participants to be ready to commit. Participants acknowledge and hold the commit in a pending state.
2. Phase 2 — Commit: Once all participants are ready, the coordinator broadcasts the decision. Participants apply the commit.

Kafka's variant uses durable write to the coordinator log (in __transaction_state) as the decision point, and WriteTxnMarker as the commit action. This is more robust than classic 2PC because the coordinator log is replicated (by default, 3 replicas), and the participants (partition leaders) do not need to hold locks.

The Coordinator as the Commit Log

The __transaction_state topic is a compacted topic with 50 partitions (default). Each partition is owned by the broker that is the leader for that partition. That broker is the Transaction Coordinator for all transactional.id values that hash to that partition.

The topic is compacted: each transactional.id has at most one active record at a time. Old records are garbage-collected as the log cleaner removes records whose keys match completed transactions.

State Machine

The transaction coordinator implements a state machine with these states:

• Empty: No transaction has been started with this transactional.id yet
• Ongoing: A transaction is in progress — partitions have been added, records written
• PrepareCommit: The coordinator has decided to commit and written this to the log. This is the point of no return — even if the coordinator crashes, recovery will complete the commit.
• CompleteCommit: All markers have been written to all participating partitions; the transaction is done
• PrepareAbort: Similar to PrepareCommit, but for abort
• CompleteAbort: Abort markers written; transaction is done

The critical invariant: once PrepareCommit is durable, the coordinator must complete the commit even if every broker in the cluster crashes and recovers. The recovery procedure reads the __transaction_state log and replays any incomplete transactions, writing the appropriate markers to data partitions.

The WriteTxnMarker Protocol

After the coordinator writes PrepareCommit to __transaction_state, it sends a WriteTxnMarker request to every partition leader that received data in the transaction. The marker request carries: (transactionalId, PID, epoch, txnResult=COMMIT|ABORT).

The partition leader writes the marker as a special control record in the log at the offset of the last record in the transaction. It then updates the transaction index (the abort index) to mark the range of records as committed or aborted. This marker is what allows read_committed consumers to filter out aborted records.

If a partition leader is unavailable when the coordinator sends the marker, the coordinator retries indefinitely (or until transaction.timeout.ms expires on the coordinator side, triggering an abort). This is why the replication factor for the transaction state topic and the data topics must be sufficient to tolerate broker failures.

The Offset Topic as a Two-Phase Participant

__consumer_offsets is just another Kafka topic. The offset commit is a transactional record written to the appropriate partition of __consumer_offsets, just like any other data write. The WriteTxnMarker protocol handles it the same way as data topic partitions.

When a consumer with isolation.level=read_committed fetches, the fetcher skips records that are in open transactions (between LSO and the high water mark). It also checks the abort index for each partition to determine whether any records in the fetch window are part of aborted transactions. Aborted records are dropped from the returned record set.

Transaction Timeout and max.poll.interval.ms Interaction

These two settings interact in a way that can silently break exactly-once if not carefully tuned. transaction.timeout.ms limits how long a transaction can be open on the coordinator side, while max.poll.interval.ms limits how long the consumer can go between polls.

The Conflict

In the read-process-write loop, the transaction must be opened before processing and closed after. If processing takes longer than max.poll.interval.ms, the consumer group coordinator triggers a rebalance — but the old consumer instance is still alive (it's in the middle of processing). This is the classic "stuck consumer" problem.

When the rebalance revokes the partition from the old instance and assigns it to a new instance, the new instance starts consuming from the last committed offset. If the old instance had already produced some records in its open transaction, those records are now orphaned — the old instance may still try to call commitTransaction() (which will succeed but write to a partition the new instance also owns, creating duplicates) or may be fenced.

The Fencing Outcome

With transactional.id set, the old producer instance is fenced when it tries to produce or commit. The coordinator bumps the epoch on initTransactions() by the new instance, so the old producer's next produce request gets ProducerFencedException. This is the correct behavior — the new instance's transaction is the authoritative one.

But the orphaned transaction's records may have been visible to a read_uncommitted consumer before fencing occurred. With read_committed, those records are hidden until the orphaned transaction is either committed or aborted. The coordinator's background expiry task will eventually abort it (if it exceeds transaction.timeout.ms), but until then, the transaction appears "open" and holds the LSO back.

Recommended Configuration

// Consumer side
props.put("max.poll.interval.ms", 300000); // 5 minutes — generous processing window
props.put("session.timeout.ms", 45000);    // group coordinator heartbeat timeout
props.put("heartbeat.interval.ms", 15000);  // must be < session.timeout.ms / 3

// Producer side
props.put("transaction.timeout.ms", 60000); // must be < max.poll.interval.ms

// Rule: transaction.timeout.ms should be at least 2x your expected processing time
// Rule: max.poll.interval.ms should be at least 3x your expected processing time
// to give headroom for GC pauses, network jitter, and rebalance delays

When to Increase max.poll.interval.ms

The max.poll.interval.ms default of 5 minutes is usually sufficient for batch processing jobs that process tens of thousands of records per poll. But for heavy processing (ML inference, database joins, external API calls), you may need to increase it. The cost of a large max.poll.interval.ms is slower rebalance response — if a consumer instance dies, it takes longer for the group to detect the death and reallocate partitions.

If your processing involves external calls with variable latency, consider wrapping them in a circuit breaker and failing fast rather than timing out. This keeps the transaction short and the poll cycle healthy.

Making Non-Idempotent Operations Idempotent

Kafka EOS guarantees that a record is delivered exactly once to the consumer. But if the consumer's processing involves a non-idempotent side effect (e.g., sending an email, incrementing a counter, triggering a webhook), that side effect can happen multiple times. Idempotency must be implemented at the application layer.

Deduplication via Record Keys

The simplest idempotency strategy: use the record key as a deduplication token. If the record key is a stable, unique identifier for the operation (e.g., a payment ID, order ID, or event UUID), the consumer can check whether that key has been processed before applying the side effect.

• In-memory set: Track recently processed IDs in a ConcurrentHashSet. Simple, but loses state on restart.
• Redis: SETNX with TTL for distributed deduplication. Fast but adds a dependency.
• Database: Upsert with the key as primary key. If the key already exists, the upsert is a no-op.

Idempotency Keys for External APIs

Many external APIs (payment gateways, SMS providers, webhooks) support idempotency keys. The pattern: when calling an external service, include a stable idempotency key in the request (e.g., the Kafka record offset + partition as a composite key, or the transactional.id + epoch + sequence). If the API supports idempotent replay, it will deduplicate the request server-side.

Stripe's idempotency key is a well-documented example: include Idempotency-Key: order-123-attempt-1 in the request header, and Stripe guarantees that replaying the same request within 24 hours produces the same response without double-charging.

// Idempotent external API call
String idempotencyKey = record.topic() + "-" + record.partition() + "-" + record.offset();
HttpRequest request = HttpRequest.newBuilder()
    .uri(URI.create("https://api.paymentprovider.com/charges"))
    .header("Authorization", "Bearer " + apiKey)
    .header("Idempotency-Key", idempotencyKey)
    .POST(HttpRequest.BodyPublishers.ofString(jsonPayload))
    .build();

// The payment provider deduplicates by idempotency key
HttpResponse<String> response = httpClient.send(request,
    HttpResponse.BodyHandlers.ofString());

if (response.statusCode() == 200) {
  processPaymentSuccess(response.body());
} else if (response.statusCode() == 409) {
  // Already processed (idempotent replay returned original result)
  log.info("Duplicate charge attempt detected: {}", idempotencyKey);
} else {
  throw new RuntimeException("Unexpected response: " + response.statusCode());
}

幂等性钥匙 (Idempotency Keys) in Distributed Systems

For systems that don't natively support idempotency keys, there are patterns to make operations idempotent:

• Command pattern with version vectors: If your operation is "set value X to new state Y," use the current state as a precondition: UPDATE ... WHERE current_state = expected_state. If the WHERE clause doesn't match, the update is a no-op — someone else updated it first.
• Optimistic locking: Add a version column to your entity. Update only where version = expected_version, then increment version. The operation succeeds exactly once, and concurrent updates fail and trigger retry with the new version.
• saga pattern: For multi-step workflows where each step is idempotent but the overall sequence is not, use a saga with compensating transactions. If step 3 fails, steps 1 and 2 are undone via compensating actions. This is the standard approach for distributed business transactions.

The Idempotency Window vs. the Exactly-Once Window

Kafka's idempotency window covers retries within a single producer instance's lifetime (until JVM restart). EOS covers producer restarts via transactional.id. But neither covers the case where a consumer processes a record, commits the offset, then crashes before the side effect completes. On restart, the consumer resumes after the committed offset — the record is not redelivered, but the side effect never happened.

The fix: always design for side effects to be applied before the offset is committed, and use the transactional boundary to wrap both. If the side effect must be applied after the Kafka transaction commits (e.g., sending a notification that the data is committed), use an outbox pattern where the notification is written to a Kafka topic in the same transaction, and a separate relay writes it to the notification system.

Kafka Streams EOS Internals: Test Driver, State Stores, Changelog

Understanding the internals of how Kafka Streams achieves exactly-once helps when debugging production issues, tuning performance, and writing accurate tests.

The StreamsTask and Its Transactional Producer

Each Kafka Streams task (StreamsPartitionManager.SupervisorStreamTask) is assigned a dedicated KafkaProducer instance initialized with a transactional.id derived from the application ID, topology name, and task ID: $applicationId-$topologyName-task-$taskId. This means each task has its own fencing identity and its own transaction state partition.

The task lifecycle:

1. Initialize: producer.initTransactions() — contacts the coordinator for this task's transactional.id, gets the current epoch, fences any zombie.
2. Poll: Consumer returns records. The task buffers them.
3. Process: Records are applied to the state store (e.g., RocksDB). Each state update is also queued as a changelog write.
4. Produce: producer.beginTransaction() opens the transaction. All output records (to sink topics) and changelog records are queued via producer.send().
5. Commit: producer.sendOffsetsToTransaction() adds the consumer offset commit to the transaction. producer.commitTransaction() closes and commits atomically.

State Store Changelog and EOS

A state store with logging enabled (.withLoggingEnabled()) writes all changes to a compacted changelog topic. With EOS, these writes are part of the task's transaction. The changelog topic's retention is set long enough to support recovery: by default, cleanup.policy=compact, so the changelog retains the latest value per key indefinitely.

The EOS guarantee for state stores: if a task fails and restarts, its state store is rebuilt by re-consuming the changelog from the beginning (or from the last checkpointed offset). Because the changelog consumption and the consumer offset commit are in the same transaction, the rebuilt state is consistent with the consumer position.

Performance implication: with EOS, each state store update generates a Kafka record to the changelog topic in the same transaction as the output. This triples the write pressure: one record for the output topic, one for the changelog, and one for the offset commit. For high-state topologies, this can be a significant throughput hit.

Checkpoint Files

Kafka Streams writes checkpoint files (in the application state directory, typically on local disk) that store the mapping from partition to offset for each source topic. These are used during recovery to know where to seek in each topic's partition. With EOS, the checkpoint file is updated after commitTransaction() succeeds, so it always reflects the last committed offset.

Topology Test Driver with EOS

The TopologyTestDriver simulates the EOS transaction lifecycle using in-memory state. It does not need a Kafka cluster — the test driver intercepts send() calls and routes them to an in-memory "topic store" that mimics the transactional behavior: writes are buffered until commitTransaction(), and aborted transactions discard their buffered writes.

This allows you to write unit tests that verify exactly-once behavior:

• Process a record, verify output appears only after commit
• Simulate an exception, verify abort discards output and offsets
• Verify that a second pipeInput() with the same key does not duplicate state store entries

Exactly-Once in Connect: IdempotentProducer and transactional.id

Kafka Connect is the integration layer for moving data between Kafka and external systems. The Connect framework has its own EOS implementation that differs from the plain producer transactions API.

Connect's Internal Producer

When a Connect worker runs in distributed mode with EOS enabled, each task uses a dedicated IdempotentProducer wrapper around the Kafka producer. The IdempotentProducer is configured with a transactional.id that is derived from the connector's worker ID and task ID. This allows the worker to fence zombie tasks without requiring the Connect framework to manage epoch bumping manually.

Connect EOS Configuration

# worker.properties for Connect cluster with EOS
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092

# Enable exactly-once for all connectors on this worker
# Valid values: exactly_once (v1), exactly_once_v2 (v2, default)
producer.enable.idempotency=true
producer.transactional.id=connect-$HOSTNAME-$CONNECTOR_NAME-$TASK_ID

# Consumer side: read_committed for source connectors that consume from Kafka
consumer.isolation.level=read_committed

# Transaction timeout — must be less than consumer max.poll.interval.ms
producer.transaction.timeout.ms=60000

# Exactly-once for sink connectors: the connector will use
# producer.transactional.id and commit offsets transactionally
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3

Source Connectors and EOS

A source connector ingests data from an external system into Kafka. The connector generates records and passes them to the Connect framework's producer. With EOS, the Connect framework wraps each batch of source records in a transaction. If the connector restarts mid-batch, the partially-written transaction is aborted, and the next batch starts from a clean position. The external source's offset (e.g., the database SCN or file position) is stored in the Kafka offset topic as part of the transaction, making the offset commit atomic with the data write.

Sink Connectors and EOS

A sink connector consumes from Kafka and writes to an external system. With EOS, the Connect framework manages the read-process-write loop: each task has a transactional producer, and the offset commit is atomic with the output write. The sink connector must implement idempotent writes to the external system.

The recommended pattern for sink connectors: use the record's topic-partition-offset as an idempotency key when writing to the external system. Many databases support this via upsert semantics. If the database doesn't support upsert, the connector can write to a staging table in the same transaction and use a merge operation.

The Connector Client Config Override

Connect allows individual connectors to override the worker's default producer/consumer configuration. This is useful when different connectors have different EOS requirements:

// Per-connector override in connector config
{
  "name": "jdbc-sink-connector",
  "config": {
    "connector.class": "JdbcSinkConnector",
    "tasks.max": "4",
    "topics": "orders",
    "connection.url": "jdbc:postgresql://db:5432/orders",
    "pk.mode": "record_key",
    "pk.fields": "order_id",
    "insert.mode": "upsert",
    // Override the worker's transactional config for this connector
    "producer.override.transactional.id": "jdbc-sink-orders-$TASK_ID",
    "producer.override.enable.idempotence": "true",
    "consumer.override.isolation.level": "read_committed"
  }
}

The Idempotency Window: How Long Kafka Tracks Producer IDs

The idempotency window is the time and sequence space that the broker tracks per (PID, partition) to deduplicate retries. Understanding its limits helps you diagnose duplicate records in production.

What the Broker Tracks

The broker's ProducerStateManager (in the broker's log manager) maintains a cache of producer states. For each (PID, partition), it stores: (lastSeq, lastEpoch, currentProducerId). When a batch arrives with a sequence number, the broker checks whether it falls within the deduplication window.

The cache has a maximum size controlled by max.produce.requests.catches. When the cache is full, the least-recently-used PID entries are evicted. If an evicted PID's producer sends a late retry after the eviction, the broker has no record of the previous sequence, and the retry is accepted as a new record.

For typical workloads, this is not a concern: the cache eviction happens over a much longer timescale than the typical retry window (which is bounded by request.timeout.ms). But for very high-partition-count deployments with many producers, it is possible to overflow the cache.

Sequence Number Space

The sequence number is a 32-bit signed integer (int32). It starts at 0 for a new PID and increments with each batch. It can wrap around after 2^31 batches, at which point the producer must call InitProducerId again to get a fresh PID. This is a rare event for normal throughput but possible for producers that have been running for years at very high volume.

The transactional.id.expiration.ms Setting

transactional.id.expiration.ms (default: 7 days) controls how long the Transaction Coordinator retains a transactional.id's state after the last transaction. This is separate from the producer state cache — it applies to the __transaction_state topic. If a producer with a given transactional.id goes silent for more than 7 days, the coordinator removes its metadata from the log, and the next time that transactional.id calls initTransactions(), it starts with epoch 0 again (no zombie fencing needed, since there are no zombies after 7 days).

This setting also controls when zombie producers are considered dead. A producer with epoch N is fenced by any new instance with epoch N+1. But if the original producer was silent for 7+ days, its epoch record is expired, and the new producer starts fresh.

Diagnosing Duplicates Caused by Idempotency Window Overflow

If you see duplicate records with the same key in your output topic, even with idempotence enabled, the likely causes are:

• Cache eviction: Too many PIDs are active simultaneously, overflowing the producer state cache. Symptoms: a producer generates duplicates after running for a while, especially under memory pressure. Fix: increase max.produce.requests.catches on the broker.
• Sequence wraparound: A producer has sent 2^31 batches on a single partition. The sequence number wraps and the broker sees it as a new sequence. Fix: restart the producer to get a new PID.
• Producer restart mid-flight: A producer had 5 batches in flight (at the idempotence limit), restarted, and the new PID sends a batch with sequence 0. The old batches (still in flight from the previous PID) arrive after the restart and are deduplicated as expected. But if the broker evicted the old PID's state during the restart window, the late-arriving old batches are written as new records.

Exactly-Once Comparison: Pulsar, RabbitMQ, Kinesis

Each message streaming system makes different tradeoffs for exactly-once semantics. Understanding the differences helps when evaluating systems or debugging cross-system issues.

Apache Pulsar

Apache Pulsar implements exactly-once via its transactional API (introduced in Pulsar 2.10). Pulsar's model is similar to Kafka's: transactions span multiple acknowledgments and produce requests, using an internal topic (__transaction_event) as the commit log. Key differences:

• Pulsar's transactions are coordinated by the Transaction Coordinator, a service that runs in each broker rather than being colocated with a partition leader.
• Pulsar supports multi-topic transactions that can span writes to topics across different tenants and namespaces in a single transaction.
• Pulsar uses a two-phase commit with a TCP-based protocol for the marker writes, rather than Kafka's WriteTxnMarker RPC.
• The producer ID is called producerName in Pulsar, and fencing is based on version numbers rather than epochs.
• Pulsar's schema system has its own transaction integration — schema mutations can be included in transactions, ensuring schema and data are committed atomically.

RabbitMQ

RabbitMQ does not support exactly-once in the Kafka sense. Its at-least-once with manual ack is the strongest guarantee. RabbitMQ uses acks (publisher confirms) and consumer acks (manual acknowledgment) as separate mechanisms:

• Publisher confirms: the broker acknowledges that a message was written to disk. With confirms, RabbitMQ provides at-least-once for producers (duplicate on retry).
• Consumer acks: the consumer sends an ack after processing. If the consumer dies before acking, the message is redelivered (at-least-once).
• RabbitMQ has a dead-letter exchange for failed messages, but this is a separate transport, not part of the message semantics.
• To achieve exactly-once in RabbitMQ, you need a database transaction that wraps both the business logic and the rabbitmq ack — but RabbitMQ doesn't support XA, so this is at-best-once semantics from the system's perspective.
• RabbitMQ's mirrored queues (classic mirrored queues, not Quorum queues) have their own failure modes around synchronization, and messages can be lost during leader failover.

Amazon Kinesis

Amazon Kinesis Data Streams provides at-least-once by default. Exactly-once is achieved through application-level deduplication:

• Sequence numbers + Iterator: Each record in Kinesis has a sequence number assigned by the shard iterator. A consumer can store the sequence number of the last processed record and start from there on restart — but if the processing happened but wasn't committed before crash, the record is reprocessed.
• Enhanced fan-out: Kinesis Data Streams enhanced fan-out uses a separate pipe per consumer, with independent delivery guarantees. Duplicate delivery is possible during consumer failovers.
• No native transaction API: Kinesis does not have a native exactly-once API equivalent to Kafka's transactions. The recommended pattern is application-level deduplication using a unique key (e.g., event ID) stored in DynamoDB with a TTL.
• Kinesis Data Analytics: Provides exactly-once processing for its SQL and Flink-based runtimes, but this is scoped to the Kinesis Analytics service, not the raw Kinesis Streams API.

Comparison Table

Transaction Guarantees and Fencing: Producer Epoch and Zombie Blocking

Fencing is the mechanism that prevents a zombie producer — an old instance that was believed dead but is still alive on the network — from writing conflicting records. It is the cornerstone of Kafka's exactly-once guarantee for multi-instance processors.

The Zombie Producer Problem

In a distributed system with a orchestrator (Kubernetes, Marathon, etc.) that restarts instances based on health checks, a producer instance can be in three states simultaneously:

1. Alive: The JVM is running, threads are executing, TCP connections are open.
2. Perceived dead by the orchestrator: The health check failed (e.g., the JVM was paused during GC) or the orchestrator killed the old pod before the new one was ready.
3. Perceived dead by Kafka: The broker has not heard from the producer within the session.timeout.ms or the producer's TCP connection dropped.

Without fencing, state 2 and state 3 overlapping is catastrophic: two instances with the same logical identity are both producing to Kafka, one writing offsets that conflict with the other's. Downstream consumers see duplicate records (from the same offset being produced twice), or worse, records from the two instances interleave in the output.

Epoch Bumping: The Fencing Protocol

When a producer with transactional.id calls initTransactions(), the coordinator performs these steps:

1. Look up the current producerEpoch for this transactional.id in __transaction_state
2. If found, increment it by 1 (or start at 0 if new)
3. Write the new epoch to __transaction_state
4. Return the new epoch to the producer

Every produce request then carries (transactionalId, PID, epoch). The partition leader checks: if the incoming epoch is less than the coordinator's current epoch for this transactional.id, the request is rejected with ProducerFencedException. The old producer cannot write any more records.

The Interaction Between PID and Epoch

The PID and epoch serve different roles in the fencing system:

• PID is assigned by the broker's InitProducerId and is used by the broker's idempotent producer dedup logic. It is opaque and changes on each JVM restart.
• Epoch is managed per transactional.id in __transaction_state. It increments on each initTransactions() call, regardless of whether the PID changes.

A producer that has never set transactional.id has no epoch — it is tracked only by PID. When it restarts, it gets a new PID and the old PID's in-flight requests are orphaned and eventually timed out by the broker. This is acceptable for at-least-once workloads where duplicates are filtered by the consumer.

A producer that sets transactional.id is fenced by epoch even if the PID is the same. This is why epoch bumping is the critical operation — it provides stability of identity across PID changes (JVM restarts).

Static Membership and Fencing

Kafka consumer groups support static membership via group.instance.id. With static membership, when a consumer instance leaves the group (voluntarily or due to a health check failure), it is given a grace period (group.initial.rebalance.delay.ms) before the group rebalances. If the instance rejoins within the grace period with the same group.instance.id, it gets the same partitions assigned without a rebalance.

In the context of EOS with static membership, the group.instance.id is included in the ConsumerGroupMetadata that is passed to sendOffsetsToTransaction(). The coordinator validates that the offset commit is coming from the legitimate owner of the partition. If a rebalance has already occurred and another instance owns the partition, the stale offset commit from the static member is rejected.

Strict Ordering of Fenced Requests

The broker processes produce requests sequentially per partition from the same producer (due to the in-flight window of 5). The epoch check happens at the broker, not at the coordinator. This means:

• The old producer's requests are rejected one by one as they arrive at the partition leader
• The old producer gets a ProducerFencedException back on the first fenced request
• The old producer should immediately call producer.close() and exit

The new producer's initTransactions() call is idempotent — calling it multiple times with the same transactional.id just returns the current epoch. So the new instance can safely call initTransactions() before starting to produce.

What the Coordinator Guarantees

The Transaction Coordinator provides these guarantees:

• At most one producer with a given transactional.id can make progress at any time — the zombie is fenced by epoch.
• Once PrepareCommit is durable, the transaction will complete — the coordinator's log is the commit log, and the recovery procedure replays incomplete transactions.
• Transaction state is durable because the __transaction_state topic is replicated (default RF=3) and compacted.

What the coordinator does not guarantee:

• Latency — if the coordinator broker is slow, all transactions for its partition are delayed
• Progress if the coordinator broker is down — unless the cluster has sufficient replicas for the coordinator's topic partition to maintain quorum
• Exactly-once delivery to non-Kafka systems — this requires application-level idempotency

Two-Phase Commit Protocol: Full Annotated Trace

A complete annotated trace of a multi-partition transaction with two output partitions and an offset commit, including the coordinator state transitions at each step.

// Setup
transactional.id = "orders-processor-1"
transaction.timeout.ms = 60000
partitions involved: orders-enriched-0, orders-enriched-1, __consumer_offsets-12
coordinator: broker-2 (partition of __transaction_state for this transactional.id)

=== PHASE 1: Initialization ===

Producer -> Broker-2 (TxnCoord): InitProducerId(transactional.id="orders-processor-1")
  Broker-2 looks up transactional.id in __transaction_state:
    NOT FOUND → assign new PID=1003, epoch=1
  Broker-2 writes: Empty -> Ongoing (PID=1003, epoch=1) in __transaction_state
  Broker-2 -> Producer: InitProducerIdResult(PID=1003, epoch=1)
  Producer stores: PID=1003, epoch=1 locally

=== PHASE 2: Adding Partitions ===

Producer -> Broker-2 (TxnCoord): AddPartitionsToTxn([orders-enriched-0, orders-enriched-1])
  Broker-2 updates __transaction_state:
    Ongoing(topicPartitions=[orders-enriched-0, orders-enriched-1])
  Broker-2 -> Producer: success

=== PHASE 3: Producing to Data Partitions ===

Producer -> Broker-4 (Leader of orders-enriched-0):
  ProduceRequest(transactional=true, PID=1003, epoch=1, seq=0..N, data=[...])
  Broker-4 appends records to log, records lastSeq=0+N
  Broker-4 -> Producer: baseOffset=5001

Producer -> Broker-5 (Leader of orders-enriched-1):
  ProduceRequest(transactional=true, PID=1003, epoch=1, seq=0..M, data=[...])
  Broker-5 appends records to log, records lastSeq=0+M
  Broker-5 -> Producer: baseOffset=3001

=== PHASE 4: Offset Commit (part of transaction) ===

Producer -> Broker-2 (TxnCoord):
  AddOffsetsToTxn(__consumer_offsets-12, group=orders-group)
Producer -> Broker-2 (TxnCoord):
  TxnOffsetCommit(group=orders-group, offsets=[
    (orders-0, offset=2001),
    (orders-1, offset=1501)
  ])
  Broker-2 writes a transactional record to __consumer_offsets-12:
    Record(transactionalId="orders-processor-1", PID=1003, epoch=1,
           group=orders-group, offsets=[...])
  Broker-2 updates __transaction_state:
    Ongoing(topicPartitions=[orders-enriched-0, orders-enriched-1, __consumer_offsets-12])

=== PHASE 5: End Transaction ===

Producer -> Broker-2 (TxnCoord): EndTxn(commit=true)

  Coordinator state transition: Ongoing -> PrepareCommit
  Broker-2 writes PrepareCommit record to __transaction_state (durable)
  *** DECISION POINT: Transaction is now decided. Coordinator must complete. ***

  Broker-2 -> Broker-4 (Leader orders-enriched-0): WriteTxnMarker(COMMIT, PID=1003, epoch=1)
  Broker-4 writes control record at baseOffset=5001+N: COMMIT marker for PID=1003
  Broker-4 updates abort index: txn PID=1003, result=COMMIT, range=[5001..5001+N]

  Broker-2 -> Broker-5 (Leader orders-enriched-1): WriteTxnMarker(COMMIT, PID=1003, epoch=1)
  Broker-5 writes control record at baseOffset=3001+M: COMMIT marker for PID=1003
  Broker-5 updates abort index: txn PID=1003, result=COMMIT, range=[3001..3001+M]

  Broker-2 -> Broker-6 (Leader __consumer_offsets-12): WriteTxnMarker(COMMIT, PID=1003, epoch=1)
  Broker-6 writes COMMIT marker to __consumer_offsets-12

  Coordinator state transition: PrepareCommit -> CompleteCommit
  Broker-2 writes CompleteCommit record to __transaction_state

=== OBSERVER EFFECT ===

Consumer with read_committed:
  Before: LSO for orders-enriched-0 = 5001 (open txn holds it back)
          LSO for orders-enriched-1 = 3001 (open txn holds it back)
  After:  LSO for orders-enriched-0 = 5001+N+1 (txn committed, records now visible)
          LSO for orders-enriched-1 = 3001+M+1 (txn committed, records now visible)
          __consumer_offsets-12 committed at (orders-0→2001, orders-1→1501)

Consumer Side: isolation.level and the Last Stable Offset

The isolation.level config controls what a consumer sees during a transaction. It is the final gate that filters aborted records and provides the consistent view that exactly-once depends on.

read_uncommitted (Default)

With read_uncommitted, the consumer fetcher returns all records as soon as they are written to the partition leader, regardless of their transaction state. Aborted records are visible. This is the fastest mode because there is no buffering — records flow as soon as they are acknowledged by the leader.

Use read_uncommitted when:

• You don't use transactions at all (at-least-once pipeline)
• Your consumers can filter aborted records themselves (e.g., by checking the transaction marker in the record header)
• Latency is more important than consistency (real-time dashboards, monitoring)

read_committed

With read_committed, the consumer fetcher tracks the Last Stable Offset (LSO) for each partition. The LSO is the offset of the first record that is part of an open (non-committed, non-aborted) transaction. All records at offsets ≥ LSO are buffered in the fetcher.

When a transaction commits:

1. The coordinator writes a COMMIT marker to the partition
2. The partition leader updates the abort index to mark the range as committed
3. The consumer fetcher sees the marker, advances the LSO past the committed transaction's end
4. The buffered records are released to the consumer poll result

When a transaction aborts:

1. The coordinator writes an ABORT marker to the partition
2. The abort index is updated
3. The consumer fetcher drops the aborted records from the buffer

The Abort Index

The abort index is an in-memory index in the log segment manager that maps (PID, txnStartOffset, txnEndOffset) to the outcome (commit or abort). It is built lazily as records are fetched by read_committed consumers. The index is stored in the log segment's .index.transaction file and is used to quickly determine whether any records in a fetch window belong to aborted transactions.

Performance Implications of read_committed

The main cost of read_committed is the LSO boundary: if a transaction is open for longer than your poll interval, all consumers with read_committed will see no new records for that partition until the transaction closes. This is the "LSO stall" problem.

The LSO stall manifests as: your consumer poll returns empty results even though records are being produced, because they are in an open transaction. The stall duration is bounded by the longest open transaction's lifespan.

Mitigation strategies:

• Keep transactions short — process and commit within max.poll.interval.ms
• Use small batch sizes to reduce per-record latency in the transaction
• Monitor the LSO lag metric (consumer.lag is the lag from the high water mark, while LSO is the lag from the open transaction boundary)
• If latency is critical, consider read_uncommitted with application-level filtering

Tradeoffs: When Exactly-Once Is and Isn't the Right Choice

Exactly-once is not always the right choice. It has real costs in throughput, latency, and operational complexity. Evaluate honestly before enabling it.

Throughput Overhead

Transactions add overhead at every stage of the pipeline:

• Producer side: Each record batch is held in the producer's transaction buffer until commitTransaction(). With exactly_once_v2, batching is improved, but there is still overhead from the transaction lifecycle management.
• Coordinator side: The Transaction Coordinator processes one extra RPC per transaction (EndTxn) plus one WriteTxnMarker per partition involved. For a transaction with 10 partitions, this is 11 extra RPCs per transaction.
• Consumer side: The fetcher maintains an LSO tracker per partition, and aborted records are filtered via the abort index. This adds CPU overhead on each fetch.
• Broker disk: Transaction markers are written to every partition that participates in a transaction. For high-throughput pipelines, this increases log segment churn.

Empirical numbers: at LinkedIn's Kafka deployment, enabling exactly-once introduced approximately 10–30% throughput overhead compared to at-least-once, depending on the transaction size and partition count. This is an acceptable cost for correctness-sensitive pipelines but a poor tradeoff for high-volume, dedupliable workloads.

Operational Complexity

EOS adds these operational concerns:

• Transaction state topic monitoring: The __transaction_state topic's health and lag must be monitored. If the coordinator for a given transactional.id is down, no producer with that ID can make progress.
• Epoch exhaustion: Rare but catastrophic — if the epoch wraps, the transactional.id is effectively dead for 7 days (until transactional.id.expiration.ms expires it).
• LSO staleness: Monitoring the LSO lag per partition to detect stalled transactions.
• Abort index growth: The abort index grows with the number of aborted transactions. If many transactions abort (e.g., due to validation errors mid-transaction), the index can grow large.

When At-Least-Once Is Sufficient

If your downstream consumers can safely deduplicate by key, at-least-once is simpler and faster. Examples:

• Log aggregation: If the same log line is delivered twice, deduplication by (timestamp, hostname, sequence) is trivial.
• Metrics pipelines: If the same metric reading is applied twice to a timeseries DB with upsert, the second write is idempotent.
• Clickstream: A duplicate click event is harmless if the analytics query uses approximate algorithms (HyperLogLog, Count-Min Sketch).
• ETL to data lake: S3 writes are immutable files. Re-writing the same file with the same content is safe. File naming by offset prevents duplicates.

Frequently Asked Questions