Spring Kafka Streams Exactly-Once Semantics and Stripe Integration: How StreamsUncaughtExceptionHandler REPLACE_THREAD Re-processes the Uncommitted Batch from the Last Checkpoint Generating UUID_B While ch_A Stays Committed on Stripe, Processor.process() KeyValueStore Guard Is Rolled Back Atomically with the Consumer Offset — Re-delivery Finds Empty Guard and Creates ch_B, and TaskMigratedException During Cooperative Rebalance Aborts the EOS Transaction Before Stripe Evidence Is Durable — New Task Owner Calls Stripe with UUID_B

Three Spring Kafka Streams EOS Stripe idempotency failure modes, each mechanically distinct from all prior posts in this series. This post covers Kafka Streams’ topology execution model and its specific EOS implementation — StreamsUncaughtExceptionHandler, the Processor API state store lifecycle, and cooperative partition rebalancing — and the three double-charge patterns that arise when UUID.randomUUID() is used as the Stripe idempotency key inside a Kafka Streams topology.

The Spring Kafka EOS post covered three Spring Kafka consumer/producer EOS failure modes: DefaultErrorHandler seek-and-re-poll, @Retryable outer / @Transactional(kafkaTransactionManager) inner AOP ordering, and ChainedKafkaTransactionManager sequential commit. The Micronaut Kafka EOS post covered three Micronaut-specific EOS failure modes: transactional producer fencing, read_committed concurrent PENDING guard race, and sendOffsetsToTransaction() atomicity. This post covers three failure modes specific to Kafka Streams’ topology execution layer. The root cause in all three is the same: the Kafka Streams EOS guarantee covers only Kafka-internal state (output topic records, state store changelog records, consumer offset advance). It does not extend to external HTTP calls to Stripe. When a Kafka Streams task re-processes a record — for any reason — the processor code runs again from the beginning, and UUID.randomUUID() at the beginning of the processing path generates a new UUID that Stripe has never seen.

The Kafka Streams EOS guarantee: what it covers and what it does not

Kafka Streams with processing.guarantee = exactly_once_v2 (EOS v2, requiring Kafka brokers 2.5+) provides the following guarantee: for a given input record in a source topic, its effects on Kafka-internal state are reflected exactly once across the output topics and state store changelog topics. This is implemented via Kafka producer transactions: the StreamThread bundles the output records, state store changelog writes, and consumer offset advance for a batch of input records into a single Kafka producer transaction, and either commits all of them atomically or aborts all of them.

What EOS v2 does not guarantee:

• That the processor code (the lambda passed to foreach(), map(), flatMapValues(), or the Processor.process() implementation) executes exactly once per input record. Processor code can execute multiple times; only its committed Kafka output is deduplicated.
• That external side effects (HTTP calls, JDBC writes to an external database, Stripe charges) are committed or rolled back atomically with the Kafka transaction.
• That a record that was processed by a failed StreamThread will not be re-processed by a replacement StreamThread or a new task owner after a rebalance.

These limitations are not bugs in Kafka Streams. They are inherent to the architecture: Kafka Streams EOS is a distributed coordination protocol for Kafka-internal state. Extending it to arbitrary external systems would require distributed two-phase commit, which Kafka Streams explicitly does not implement. The consequence for developers using Stripe inside a Kafka Streams topology: the EOS guarantee does not protect against double Stripe charges. Only a correct Stripe idempotency key strategy does.

Mode 1: StreamsUncaughtExceptionHandler.REPLACE_THREAD re-processes the uncommitted batch — UUID.randomUUID() at topology entry generates UUID_B

Kafka Streams runs topology processing on one or more StreamThread instances. Each StreamThread runs a main loop: runOnce() fetches a batch of records from the assigned input partitions, processes them through the topology, and calls commitAndUpdateLocals() to commit the EOS transaction at commit.interval.ms intervals (default 100ms for EOS v2). If the StreamThread’s main loop throws an uncaught exception, Kafka Streams calls the registered StreamsUncaughtExceptionHandler.

Since Kafka Streams 2.8, the default behavior for most uncaught exceptions is StreamThreadExceptionResponse.REPLACE_THREAD: Kafka Streams starts a new StreamThread to take over the dead thread’s task assignments. The replacement thread:

1. Is assigned the same tasks (input partitions) that the dead thread held.
2. Restores each task’s state stores from the Kafka changelog topics, seeking to the last committed checkpoint position.
3. Resets the input partition consumer position to the last committed offset for each task (the offset that was committed in the last successful EOS transaction commit).
4. Re-processes all records from the last committed offset — i.e., all records in the uncommitted batch that the dead thread was processing when it threw.

The following topology places a Stripe call inside foreach(), generating UUID.randomUUID() at the beginning of the processing path:

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "billing-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

StreamsBuilder builder = new StreamsBuilder();

// Persistent KeyValueStore — changelog topic "billing-streams-app-payment-record-store-changelog"
StoreBuilder<KeyValueStore<String, ProcessingResult>> storeBuilder =
    Stores.keyValueStoreBuilder(
        Stores.persistentKeyValueStore("payment-record-store"),
        Serdes.String(),
        new ProcessingResultSerde()
    );
builder.addStateStore(storeBuilder);

builder.<String, PaymentCommand>stream("payment-commands")
    .foreach((orderId, command) -> {
        // UUID_A on first processing of this record
        String idempotencyKey = UUID.randomUUID().toString();

        // Stripe call — ch_A committed on Stripe's HTTP layer
        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(),
            command.getCustomerId(),
            idempotencyKey
        );

        // Write the processing result to the state store.
        // If this throws ProcessorStateException, the EOS transaction is not committed.
        KeyValueStore<String, ProcessingResult> store =
            streamsContext.getStateStore("payment-record-store");
        store.put(orderId, new ProcessingResult(idempotencyKey, charge.getId(), "COMPLETED"));
    });

KafkaStreams streams = new KafkaStreams(builder.build(), props);

// REPLACE_THREAD: new StreamThread takes over on uncaught exception
streams.setUncaughtExceptionHandler(exception ->
    StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD
);

Step-by-step failure trace

1. StreamThread-1 polls payment-commands, partition 0. Broker returns PaymentCommand(customerId=“cust_1”, orderId=“order_1”, amountCents=5000) at offset 42 as part of a batch of 10 records.
2. StreamThread-1 processes the batch through the topology. For the record at offset 42: UUID.randomUUID() executes at the foreach() lambda entry. idempotencyKey = UUID_A.
3. stripeClient.createCharge(5000, “cust_1”, UUID_A) → Stripe HTTP POST /v1/charges with Idempotency-Key: UUID_A. Stripe processes the charge and returns ch_A (HTTP 200). ch_A is now committed on Stripe’s side.
4. store.put(“order_1”, ProcessingResult(UUID_A, ch_A, “COMPLETED”)) → throws ProcessorStateException. The underlying RocksDB instance for the state store has encountered an I/O error (disk space exhaustion on the broker pod’s local storage). The exception propagates out of the foreach() lambda, through the topology execution chain, to StreamThread-1’s runOnce() method.
5. StreamThread-1 catches the exception in its main loop. The EOS transaction for this batch was never committed — no KafkaProducer.commitTransaction() was called. StreamThread-1 calls streamsUncaughtExceptionHandler.handle(exception). The handler returns REPLACE_THREAD.
6. Kafka Streams starts StreamThread-2. StreamThread-2 takes over the partition-0 task. It reads the state store checkpoint file from local disk (the last committed checkpoint) and restores the payment-record-store from the Kafka changelog topic up to the checkpoint position. The checkpoint does NOT include the store.put() from step 4 (that write was never committed to the changelog topic — it was part of the aborted EOS transaction).
7. StreamThread-2 resets the consumer position for partition 0 to the last committed offset for this task. Because the EOS transaction that included offset 42 (and all 10 records in the batch) was never committed, the last committed offset for partition 0 is still 42. StreamThread-2 re-polls from offset 42.
8. The broker returns the same ConsumerRecord at partition 0, offset 42. StreamThread-2 processes it through the topology. The foreach() lambda executes. UUID.randomUUID() at lambda entry generates UUID_B. This is a fresh UUID.randomUUID() call producing a value distinct from UUID_A.
9. stripeClient.createCharge(5000, “cust_1”, UUID_B) → Stripe HTTP POST /v1/charges with Idempotency-Key: UUID_B. Stripe looks up UUID_B in its 24h idempotency cache. UUID_B has never been seen. Stripe processes a new charge and returns ch_B. ch_B is committed on Stripe alongside ch_A. The customer has been charged twice.

Developer misconception

“I configured processing.guarantee = exactly_once_v2. This guarantees that each input record in payment-commands is processed exactly once. Since REPLACE_THREAD is designed to recover from StreamThread failures without data loss or duplication, and EOS is active, the replacement StreamThread will not re-process the record that already resulted in a committed Stripe charge.”

This model conflates two distinct guarantees. Kafka Streams EOS guarantees that each record’s effect on Kafka-internal state — specifically, the output topic records and state store changelog records that form the transaction payload, and the consumer offset advance — is reflected exactly once. The EOS guarantee does not cover the execution of processor code. The foreach() lambda, the map() function, the Processor.process() method — these are code paths that Kafka Streams invokes when processing a record. When the EOS transaction fails and REPLACE_THREAD re-processes the batch from the last checkpoint, the processor code executes again. Every line of the lambda executes again. UUID.randomUUID() is one of those lines.

The EOS guarantee is about the commit of results, not the execution of code. If the same record is committed twice, that is a bug (EOS prevents this). If the code that processes the record runs twice, EOS is indifferent — it only ensures the second execution’s committed output supersedes the first’s (or the first is the only one that commits). Stripe is not part of the committed output.

Fix

Replace UUID.randomUUID() with a content-hash key derived from the stable fields of the PaymentCommand. The key is computed from customerId, orderId, and amountCents — fields that are identical on every re-processing of the same logical billing event, regardless of how many REPLACE_THREAD cycles have occurred:

builder.<String, PaymentCommand>stream("payment-commands")
    .foreach((orderId, command) -> {
        // Content-hash key: same value on every REPLACE_THREAD re-processing
        // of the same PaymentCommand. SHA-256 of customerId + orderId + amountCents.
        String idempotencyKey = contentHashKey(command);

        // Guard: COMPLETED entry in state store means this record was processed
        // in a prior StreamThread run that committed its EOS transaction.
        KeyValueStore<String, ProcessingResult> store =
            streamsContext.getStateStore("payment-record-store");
        ProcessingResult existing = store.get(orderId);
        if (existing != null && "COMPLETED".equals(existing.getStatus())) {
            // Prior EOS transaction committed: state store has the entry,
            // meaning the consumer offset for this record was also committed.
            // This code path is reached only if the record was re-delivered
            // despite a committed offset — which should not happen in normal
            // operation but can occur after ApplicationId reset or partition migration.
            return;
        }

        // Stripe — content-hash key is stable on every REPLACE_THREAD re-processing.
        // Stripe returns ch_A from its 24h cache on re-processing with the same key.
        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(), command.getCustomerId(), idempotencyKey);

        store.put(orderId,
            new ProcessingResult(idempotencyKey, charge.getId(), "COMPLETED"));
    });

private static String contentHashKey(PaymentCommand command) {
    String raw = command.getCustomerId() + "|" + command.getOrderId()
                 + "|" + command.getAmountCents();
    try {
        byte[] hash = MessageDigest.getInstance("SHA-256")
                                   .digest(raw.getBytes(StandardCharsets.UTF_8));
        return Base64.getUrlEncoder().withoutPadding().encodeToString(hash);
    } catch (NoSuchAlgorithmException e) {
        throw new IllegalStateException(e);
    }
}

With this fix, StreamThread-2’s re-processing of the record at offset 42 generates the same idempotencyKey as StreamThread-1’s original attempt. Stripe looks up the key in its 24h idempotency cache, finds ch_A, and returns it without creating a new charge. No ch_B is created. The state store entry is written (inside the new EOS transaction) with ch_A’s charge ID, and the EOS transaction commits successfully.

Mode 2: Processor API KeyValueStore guard is rolled back atomically with the consumer offset — re-processing finds the guard empty and calls Stripe with UUID_B

A natural next step for a developer who understands Mode 1 is to add an idempotency guard using the Kafka Streams state store: check the store before calling Stripe; if an entry exists, skip Stripe. This guard works correctly when the EOS transaction commits. But it fails silently when the EOS transaction aborts — because the state store write and the consumer offset advance are rolled back together. The guard’s presence tracks whether the EOS transaction committed; it does not track whether Stripe was called.

The Processor API gives fine-grained control over state store access and record forwarding. The following BillingProcessor implements what appears to be a correct guard:

public class BillingProcessor
        implements Processor<String, PaymentCommand, String, PaymentResult> {

    private ProcessorContext<String, PaymentResult> context;
    private KeyValueStore<String, ProcessingRecord> store;
    private StripeClient stripeClient;

    @Override
    public void init(ProcessorContext<String, PaymentResult> context) {
        this.context = context;
        this.store = context.getStateStore("payment-record-store");
        this.stripeClient = ApplicationContext.getBean(StripeClient.class);
    }

    @Override
    public void process(Record<String, PaymentCommand> record) {
        String orderId = record.key();
        PaymentCommand command = record.value();

        // Guard: check if this orderId has already been processed.
        ProcessingRecord existing = store.get(orderId);
        if (existing != null) {
            // Already processed — forward the result without calling Stripe.
            context.forward(record.withValue(
                new PaymentResult(orderId, existing.getChargeId(), existing.getIdempotencyKey())
            ));
            return;
        }

        // Not in store — generate key and call Stripe.
        String idempotencyKey = UUID.randomUUID().toString(); // UUID_A on attempt 1

        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(), command.getCustomerId(), idempotencyKey);

        // Write to store. If this succeeds and the EOS transaction commits,
        // the guard will prevent re-processing on the next delivery attempt.
        store.put(orderId,
            new ProcessingRecord(idempotencyKey, charge.getId(), "COMPLETED"));

        context.forward(record.withValue(
            new PaymentResult(orderId, charge.getId(), idempotencyKey)
        ));
    }
}

Why the guard fails

The developer’s reasoning: “With EOS enabled, the state store write and the consumer offset advance are atomic. If the store.put() commits (EOS transaction committed), the consumer offset for this record is also advanced — the record is not re-delivered. If the store.put() is rolled back (EOS transaction aborted), the consumer offset is also rolled back — the record is re-delivered. So the guard works: if store.get() returns non-null, the prior EOS transaction committed and the consumer offset was advanced; the record is being processed again only if something unusual is happening. If store.get() returns null, the prior transaction was rolled back — Stripe was never called.”

The flaw: the reasoning correctly describes what EOS atomicity implies about the state store and the consumer offset. But it incorrectly concludes that a rolled-back state store implies the Stripe call did not happen. The sequence of events:

1. store.get(orderId) → null. Guard says: not processed yet. Proceed.
2. UUID.randomUUID() → UUID_A.
3. stripeClient.createCharge(5000, “cust_1”, UUID_A) → ch_A committed on Stripe. HTTP 200 returned.
4. store.put(orderId, ProcessingRecord(UUID_A, ch_A, “COMPLETED”)) → write issued to RocksDB. This write is buffered inside the open EOS transaction — it is NOT committed to Kafka’s changelog topic yet.
5. context.forward() → output record buffered inside the open EOS transaction on the payment-results topic — NOT committed yet.
6. The EOS transaction commit attempt fails. The cause may be a TimeoutException from the Kafka broker, a network partition, transaction.timeout.ms expiry, or any other exception during KafkaProducer.commitTransaction(). Alternatively, the StreamThread could die before the commit attempt (as in Mode 1). In any case: the EOS transaction is aborted.
7. EOS abort: the changelog write for store.put() (step 4) is rolled back — the Kafka changelog topic for payment-record-store never receives the write. On any subsequent restore, the store will not have the orderId entry. RocksDB’s in-memory representation also reverts to the pre-transaction state. The output record on payment-results (step 5) is rolled back. The consumer offset for partition 0 is not advanced.
8. Re-processing: the record at offset 42 is re-delivered to BillingProcessor.process().
9. store.get(orderId) → null. The store was rolled back. Guard says: not processed yet. Proceed.
10. UUID.randomUUID() → UUID_B. This is a fresh call producing a value distinct from UUID_A.
11. stripeClient.createCharge(5000, “cust_1”, UUID_B) → Stripe POST /v1/charges with Idempotency-Key: UUID_B. Stripe looks up UUID_B: never seen. Creates ch_B. ch_B committed alongside ch_A.

Developer misconception

“If the state store entry for orderId is absent when process() is called, then the prior EOS transaction was rolled back. EOS atomicity guarantees that if the state store write was rolled back, the consumer offset advance was also rolled back — meaning the record was re-delivered because the prior processing did not commit. If the prior processing did not commit, it means the entire unit of work was undone — including the Stripe call. A committed EOS transaction is the proof of a committed Stripe charge. An absent state store entry is the proof that no Stripe charge is committed.”

The misconception is treating the EOS transaction as a two-phase commit that includes Stripe. Kafka Streams EOS is not a distributed transaction protocol. It is a Kafka-native transactional protocol. The “unit of work” that EOS commits atomically is: output topic records + state store changelog records + consumer offset. Stripe is not a party to this transaction. Stripe processes the charge when the HTTP request arrives at its API — independently of whether the Kafka Streams EOS transaction subsequently commits or aborts. When the EOS transaction aborts, Stripe’s committed charge ch_A is not rolled back. It cannot be: Stripe has no JTA enrollment, no XA resource manager, and no hook into Kafka’s abortTransaction() protocol.

The state store guard is effective only for detecting records that are re-delivered AFTER a successful EOS commit. For records that are re-delivered because the EOS transaction aborted (the exact scenario Mode 2 describes), the state store guard is always empty on re-delivery — which is precisely when Stripe may have already committed a charge.

Fix

The fix requires an idempotency guard that survives EOS transaction aborts. A relational database row written in its own transaction (independent of Kafka Streams EOS) before the Stripe call provides this. The PENDING row commits to the database immediately — it is NOT rolled back when the Kafka Streams EOS transaction aborts. On re-processing, the database guard is found and Stripe is skipped:

@Override
public void process(Record<String, PaymentCommand> record) {
    String orderId = record.key();
    PaymentCommand command = record.value();

    // Content-hash key: stable across any number of EOS re-deliveries
    // of the same PaymentCommand.
    String idempotencyKey = contentHashKey(command);

    // External DB guard — survives EOS transaction aborts.
    // This check queries the relational DB, not the Kafka Streams state store.
    // The DB PENDING row is NOT rolled back when the EOS transaction aborts.
    Optional<PaymentRecord> existing =
        paymentRecordRepository.findByIdempotencyKey(idempotencyKey);

    if (existing.isPresent()) {
        String chargeId = existing.get().getChargeId();
        if (chargeId != null) {
            // Prior processing committed Stripe charge.
            // Forward result and let EOS commit the consumer offset.
            context.forward(record.withValue(
                new PaymentResult(orderId, chargeId, idempotencyKey)));
        }
        // If chargeId is null, a prior attempt inserted PENDING but hasn't committed Stripe.
        // Content-hash key means the Stripe call (if retried) returns ch_A from cache.
        return;
    }

    // No DB record — this is a genuine first attempt.
    // Insert PENDING row in its own transaction so it commits immediately,
    // independent of the Kafka Streams EOS transaction.
    paymentRecordRepository.insertPending(orderId, idempotencyKey);

    // Stripe — content-hash key is stable on re-delivery.
    // If this is a re-delivery after EOS abort, PENDING row was found above and
    // Stripe returned ch_A from cache. We only reach here on genuine first attempt.
    StripeCharge charge = stripeClient.createCharge(
        command.getAmountCents(), command.getCustomerId(), idempotencyKey);

    paymentRecordRepository.updateToCompleted(idempotencyKey, charge.getId());

    // Also write to Kafka Streams state store for fast in-memory lookups.
    // If this EOS transaction aborts, the state store is rolled back but
    // the DB row persists — the DB row handles re-delivery protection.
    KeyValueStore<String, ProcessingRecord> store =
        context.getStateStore("payment-record-store");
    store.put(orderId, new ProcessingRecord(idempotencyKey, charge.getId(), "COMPLETED"));

    context.forward(record.withValue(
        new PaymentResult(orderId, charge.getId(), idempotencyKey)));
}

The PENDING row’s own transaction commits to the relational database before the Stripe call. If the Kafka Streams EOS transaction subsequently aborts, the relational database row remains committed. On re-processing, paymentRecordRepository.findByIdempotencyKey() finds the PENDING row and the method returns early, skipping the Stripe call. The content-hash key ensures the same logical PaymentCommand always finds the same PENDING row, regardless of how many EOS re-deliveries have occurred.

Mode 3: TaskMigratedException during cooperative partition rebalance aborts the EOS transaction — state store evidence of Stripe call rolled back — new task owner re-processes with UUID_B

Kafka Streams 2.4 introduced the cooperative sticky assignor as the default partition assignor for KafkaConsumer-based stream processing. Cooperative rebalancing differs from eager (stop-the-world) rebalancing in one key way: partitions are revoked and re-assigned incrementally. A consumer group member can continue processing its non-revoked partitions while other partitions are being redistributed. This reduces rebalance-induced downtime significantly in large deployments.

The cooperative rebalance protocol involves two rebalance rounds for partition migrations. In the first round, the group coordinator instructs each member to revoke the partitions it should give up. In the second round, the coordinator assigns those revoked partitions to their new owners. During the first round, a Streams instance that is asked to revoke partition P continues processing its retained partitions.

The failure occurs when the StreamThread is mid-EOS-commit for a batch that includes records from partition P — the partition being revoked — at the moment the cooperative rebalance executes the revocation. The EOS commit sequence calls KafkaConsumer.commitSync() (via sendOffsetsToTransaction() and then the consumer offset commit as part of the transaction). If the partition was revoked between the last consumer.poll() and the consumer.commitSync(), the commit throws CommitFailedException. Kafka Streams wraps this as TaskMigratedException.

// Topology with KStream.transform() — uses state store and forwards to output topic
builder.<String, PaymentCommand>stream("payment-commands")
    .transform(() -> new BillingTransformer(), "payment-record-store")
    .to("payment-results");

public class BillingTransformer
        implements Transformer<String, PaymentCommand, KeyValue<String, PaymentResult>> {

    private ProcessorContext context;
    private KeyValueStore<String, ProcessingRecord> store;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
        this.store = context.getStateStore("payment-record-store");
    }

    @Override
    public KeyValue<String, PaymentResult> transform(String orderId, PaymentCommand command) {
        // Guard check
        ProcessingRecord existing = store.get(orderId);
        if (existing != null) {
            return KeyValue.pair(orderId,
                new PaymentResult(orderId, existing.getChargeId(), existing.getIdempotencyKey()));
        }

        String idempotencyKey = UUID.randomUUID().toString(); // UUID_A

        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(), command.getCustomerId(), idempotencyKey);

        // Write to state store — part of EOS transaction
        store.put(orderId, new ProcessingRecord(idempotencyKey, charge.getId(), "COMPLETED"));

        return KeyValue.pair(orderId,
            new PaymentResult(orderId, charge.getId(), idempotencyKey));
    }

    @Override
    public void close() {}
}

Step-by-step failure trace

Context: Kafka Streams application with two instances (Instance A and Instance B) and one input topic partition (partition 0). Instance A holds partition 0. A third Streams instance (Instance C) joins the consumer group, triggering a cooperative rebalance.

1. Cooperative rebalance round 1 begins. Instance A is asked to revoke partition 0 and give it to Instance B (per the assignor’s sticky rebalance calculation). Instance A receives the revocation instruction in its onPartitionsRevoked() callback, which Kafka Streams processes during the next consumer.poll() call.
2. Before Instance A processes the revocation, StreamThread-A is mid-batch. It has polled a batch of 5 records from partition 0 (offsets 100–104). For the record at offset 102: UUID.randomUUID() → UUID_A. stripeClient.createCharge(UUID_A) → ch_A committed on Stripe.
3. StreamThread-A finishes processing the batch and calls commitAndUpdateLocals() to commit the EOS transaction. The EOS commit sequence calls KafkaProducer.sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata>, groupMetadata) to include the consumer offset advance (partition 0, offset 105) in the Kafka transaction. Then it calls KafkaProducer.commitTransaction().
4. During KafkaProducer.commitTransaction(), the Kafka broker processes the commit request and attempts to advance the consumer group offset for the group. At this point, the group coordinator has already registered the revocation of partition 0 from Instance A (from the cooperative rebalance). The offset commit is rejected because the partition no longer belongs to Instance A’s generation of the consumer group. The broker returns an error code (e.g., UNKNOWN_MEMBER_ID or FENCED_LEADER_EPOCH).
5. Kafka Streams receives the EOS commit failure. It throws TaskMigratedException (wrapping the underlying CommitFailedException or producer fencing error).
6. Kafka Streams handles TaskMigratedException by aborting all active EOS transactions for the affected tasks. The abort rolls back: the state store changelog writes for records 100–104 (including the store.put(orderId, ProcessingRecord(UUID_A, ch_A, ...)) for offset 102), the output records on payment-results for records 100–104, and the consumer offset advance to 105. The consumer offset for partition 0 in the group coordinator remains at 100 (the last successfully committed offset for this partition under Instance A).
7. Cooperative rebalance round 2 completes. Instance B is assigned partition 0. Instance B’s StreamThread initializes the task for partition 0 by restoring the payment-record-store from the Kafka changelog topic up to the last committed checkpoint. The changelog does NOT have the store.put() for offset 102 (it was aborted in step 6). The consumer offset for partition 0 is 100.
8. Instance B polls partition 0 from offset 100. The broker returns records 100–104. BillingTransformer.transform() is called for offset 102. store.get(orderId) → null (the entry was rolled back in step 6). UUID.randomUUID() → UUID_B. stripeClient.createCharge(UUID_B) → Stripe POST /v1/charges with Idempotency-Key: UUID_B. UUID_B is not in Stripe’s 24h cache. Stripe creates ch_B. ch_B committed alongside ch_A. The customer has been charged twice.

Developer misconception

“I’m using cooperative rebalancing, which is designed to be a graceful, low-disruption handoff. The Kafka Streams documentation says that cooperative rebalancing reduces stop-the-world pauses by allowing incremental partition migration. Since the handoff is graceful, the StreamThread commits any in-flight work before giving up a partition. The state store is checkpointed on revocation. Any records processed between the last EOS checkpoint and the revocation are committed before the partition is handed to the new owner. The state store will have the store.put() entry on the new task owner, so the guard will prevent a double Stripe call.”

Cooperative rebalancing reduces downtime by not stopping the consumer entirely during the rebalance. It does not guarantee that in-flight EOS transactions are committed before a partition is revoked. The TaskMigratedException scenario is specifically the case where the EOS commit attempt fails because the partition has been revoked mid-commit. This is an inherent edge case in cooperative rebalancing with EOS: the rebalance and the EOS commit race, and if the rebalance wins, the EOS transaction is aborted.

The Kafka Streams documentation acknowledges this. From the Kafka documentation on TaskMigratedException: tasks that are migrated due to rebalancing will abort their active transactions and re-process from the last committed checkpoint on the new owner. This is correct behavior for EOS-internal state: the Kafka output is correct and consistent. But it is not correct behavior for external side effects that were not rolled back (because they cannot be rolled back).

The cooperative rebalance reduces the frequency of this scenario by minimizing unnecessary partition migrations. It does not eliminate the scenario. Under load, with many partitions and multiple Streams instances, TaskMigratedException during EOS commit is a recurring event, not a rare edge case.

Fix

The same two-part fix from Mode 2 applies: content-hash idempotency key and an external (non-EOS-governed) PENDING row written before the Stripe call. For the KStream.transform() pattern, the fix is cleanest when the transformer delegates the database interaction to a @Transactional Spring service bean:

public class BillingTransformer
        implements Transformer<String, PaymentCommand, KeyValue<String, PaymentResult>> {

    private ProcessorContext context;
    private KeyValueStore<String, ProcessingRecord> store;
    private BillingService billingService; // Spring-managed, @Transactional

    @Override
    public KeyValue<String, PaymentResult> transform(String orderId, PaymentCommand command) {
        // Content-hash key: stable across TaskMigratedException re-deliveries.
        String idempotencyKey = contentHashKey(command);

        // Fast path: Kafka Streams state store lookup (in-memory, nanosecond latency).
        // Present only if a prior EOS transaction committed successfully for this orderId.
        ProcessingRecord cached = store.get(orderId);
        if (cached != null) {
            return KeyValue.pair(orderId,
                new PaymentResult(orderId, cached.getChargeId(), cached.getIdempotencyKey()));
        }

        // Slow path: external DB guard — survives TaskMigratedException + EOS abort.
        // billingService.processPayment() does:
        //   1. INSERT PENDING row (own transaction, commits immediately)
        //   2. Check for existing row (concurrent guard)
        //   3. Call Stripe with content-hash key
        //   4. UPDATE row to COMPLETED with charge ID
        // The PENDING row is NOT rolled back when the Kafka Streams EOS transaction aborts.
        // On re-processing, billingService.processPayment() finds the PENDING/COMPLETED row
        // and returns the existing charge ID without calling Stripe again.
        PaymentResult result = billingService.processPayment(orderId, command, idempotencyKey);

        // Update Kafka Streams state store — rolled back with EOS on failure,
        // but that is fine: the DB row protects against the double-charge.
        // The state store provides fast-path protection for the normal commit case.
        store.put(orderId, new ProcessingRecord(
            result.getIdempotencyKey(), result.getChargeId(), "COMPLETED"));

        return KeyValue.pair(orderId, result);
    }
}

@Service
class BillingService {

    @Autowired PaymentRecordRepository repo;
    @Autowired StripeClient stripeClient;

    public PaymentResult processPayment(
            String orderId, PaymentCommand command, String idempotencyKey) {

        // This entire method runs in its own transaction context,
        // NOT in the Kafka Streams EOS transaction.
        Optional<PaymentRecord> existing = repo.findByIdempotencyKey(idempotencyKey);
        if (existing.isPresent() && existing.get().getChargeId() != null) {
            return new PaymentResult(orderId, existing.get().getChargeId(), idempotencyKey);
        }
        if (existing.isEmpty()) {
            // Insert PENDING in REQUIRES_NEW — commits immediately
            insertPending(orderId, idempotencyKey);
        }
        // Stripe — content-hash key: TaskMigratedException re-delivery gets ch_A from cache
        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(), command.getCustomerId(), idempotencyKey);
        repo.updateToCompleted(idempotencyKey, charge.getId());
        return new PaymentResult(orderId, charge.getId(), idempotencyKey);
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    void insertPending(String orderId, String idempotencyKey) {
        repo.save(new PaymentRecord(orderId, idempotencyKey, null, "PENDING"));
    }
}

After a TaskMigratedException aborts Instance A’s EOS transaction, the relational database has a COMPLETED row for idempotencyKey (with chargeId = ch_A). When Instance B re-processes the record, billingService.processPayment() finds the COMPLETED row and returns ch_A without calling Stripe. The content-hash key ensures that Instance B uses the same idempotencyKey as Instance A, so the database lookup succeeds.

Comparison table

Kafka Streams EOS scope vs. external API scope

All three modes share the same root cause: the developer treats Kafka Streams EOS as a distributed transaction protocol that covers external API calls. It does not. Understanding the precise boundary of the EOS guarantee is essential for any Kafka Streams topology that calls Stripe or any other external API:

• Inside EOS scope: Kafka output topic records (forwarded via context.forward(), KStream.to(), KTable.to(), etc.); state store changelog records (KeyValueStore.put(), SessionStore.put(), etc.); consumer offset advance for the input partitions. All of these are committed atomically in a single KafkaProducer transaction. Either all commit, or all are rolled back (abort).
• Outside EOS scope: External HTTP calls (Stripe /v1/charges); JDBC writes to a relational database unless explicitly enrolled in a distributed transaction; calls to any external service (S3 put, email send, push notification). These side effects occur when the processor code executes. They are not part of the Kafka transaction. They are not rolled back when the Kafka transaction aborts. They may occur multiple times if the processor code re-executes.

The practical consequence: any Kafka Streams topology that calls Stripe inside a processor must treat every processor invocation as a potential re-invocation. The content-hash idempotency key is the primary defense: it ensures that re-invocations of the same logical PaymentCommand produce the same Idempotency-Key header on the Stripe API call, and Stripe’s 24h idempotency cache returns the original charge instead of creating a new one. The external PENDING row (Modes 2 and 3) provides defense-in-depth: it handles the case where the content-hash Stripe call might not be safe to retry (e.g., Stripe’s idempotency cache has expired after 24 hours, or the amount has legitimately changed in a way that invalidates the content-hash key’s uniqueness assumption).

Integration test patterns for Kafka Streams EOS + Stripe

Unit tests for Kafka Streams topologies typically use TopologyTestDriver, which runs the topology synchronously and in-memory. TopologyTestDriver does not simulate EOS transaction aborts, REPLACE_THREAD, or TaskMigratedException. Testing the failure modes above requires a full embedded Kafka cluster with EOS enabled:

@SpringBootTest
@Testcontainers
class BillingTopologyEosTest {

    @Container
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.0"))
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_MIN_ISR", "1")
        .withEnv("KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR", "1");

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");

    @Autowired KafkaTemplate<String, PaymentCommand> producer;
    @Autowired KafkaStreams streams;

    // WireMock for Stripe
    static WireMockServer wireMock = new WireMockServer(WireMockConfiguration.options().dynamicPort());

    @Test
    void replaceThread_contentHashKey_stripeCalledOncePerLogicalCommand() throws Exception {
        // Arrange: produce a PaymentCommand
        PaymentCommand command = new PaymentCommand("cust_1", "order_1", 5000);
        producer.send(new ProducerRecord<>("payment-commands", "order_1", command)).get();

        // Simulate: configure WireMock to record all Stripe /v1/charges calls
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.okJson("{\"id\":\"ch_A\",\"status\":\"succeeded\"}")));

        // Simulate REPLACE_THREAD by injecting a state store write failure
        // on the first invocation (controlled by a CountDownLatch in the test)
        injectStateStoreFailureOnFirstInvocation("order_1");

        // Wait for the record to be processed (REPLACE_THREAD + re-processing)
        awaitProcessed("order_1", Duration.ofSeconds(30));

        // Assert: Stripe was called twice (once per StreamThread execution)
        // but with the SAME Idempotency-Key header both times (content-hash)
        List<LoggedRequest> stripeRequests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(2);

        String key1 = stripeRequests.get(0).getHeader("Idempotency-Key");
        String key2 = stripeRequests.get(1).getHeader("Idempotency-Key");
        assertThat(key1)
            .describedAs("Both REPLACE_THREAD executions must use same content-hash key")
            .isEqualTo(key2);
    }

    @Test
    void taskMigratedException_externalDbGuard_stripeCalledOnceForSameLogicalCommand()
            throws Exception {
        // Arrange
        PaymentCommand command = new PaymentCommand("cust_2", "order_2", 7500);
        producer.send(new ProducerRecord<>("payment-commands", "order_2", command)).get();

        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.okJson("{\"id\":\"ch_A2\",\"status\":\"succeeded\"}")));

        // Simulate: record processed to the point of Stripe call, then EOS transaction aborted
        // (mimics TaskMigratedException by directly aborting the KafkaProducer transaction
        // via a spy on the StreamThread's commit phase)
        injectEosAbortAfterStripeCallFor("order_2");

        // Wait for re-processing and successful commit
        awaitProcessed("order_2", Duration.ofSeconds(30));

        // Assert: Stripe was called twice (once per re-processing) but SAME key
        List<LoggedRequest> stripeRequests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(2);

        String key1 = stripeRequests.get(0).getHeader("Idempotency-Key");
        String key2 = stripeRequests.get(1).getHeader("Idempotency-Key");
        assertThat(key1).isEqualTo(key2);

        // Assert: only one charge committed in the DB (second Stripe call returned ch_A2 from cache)
        List<PaymentRecord> records = paymentRecordRepository.findByOrderId("order_2");
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getChargeId()).isEqualTo("ch_A2");
    }
}

The tests above assert that the Stripe Idempotency-Key header is identical on both the original processing and the re-processing path, regardless of whether re-processing is triggered by REPLACE_THREAD or by an EOS transaction abort. This is the property that Stripe’s 24h idempotency cache uses to return the original charge instead of creating a new one. Testing this property directly — rather than testing that the topology produces the correct output record — is the key difference between a test that passes in development and a test that catches the double-charge bug.