Micronaut Kafka Exactly-Once Semantics and Stripe Integration: How Producer Fencing Aborts the Kafka Transaction Leaving ch_A Committed on Stripe While the New Producer Epoch Consumer Replays the Record with UUID_B, read_committed Isolation Allows Two Concurrent Consumer Instances to Both Pass the PENDING Guard Before Either Commits a Row Creating ch_B, and sendOffsetsToTransaction() Atomicity Does Not Cover the Stripe Call — Consumer Re-reads with UUID_B After Transaction Abort

The prior Micronaut Kafka post in this series covered three failure modes at the Kafka consumer layer: @KafkaListener with @Retryable re-entering the method body via MethodInvocationContext.proceed() and regenerating the idempotency key, @KafkaClient sending non-transactionally inside a @Transactional boundary leaving the message on the broker after a JDBC rollback, and batch=true redelivering the entire batch after a mid-batch exception and re-charging already-processed records. All three failures operated at the level of Micronaut’s AOP interceptor stack and the Kafka ackMode setting. This post moves one layer deeper: to Kafka’s exactly-once semantics (EOS) APIs. EOS adds transactional producers, fencing, isolation.level=read_committed, and the sendOffsetsToTransaction() consume-transform-produce pattern. Each of these tools introduces a new surface area for Stripe idempotency failures that is invisible if you only inspect the application-layer Kafka configuration.

The three failure modes in this post are mechanically distinct from each other and from all prior Micronaut Kafka posts. The first is a fencing failure: a transactional producer calls Stripe inside its transaction boundary and is then fenced by the Kafka coordinator — abortTransaction() cancels the Kafka message but does not undo the Stripe charge, and the new producer epoch delivers the message again. The second is a partition-rebalance concurrent guard race: two consumer instances with isolation.level=read_committed process the same message in an overlapping window because the consumed offset was not committed before the rebalance, and READ COMMITTED on the payment table does not block on the first consumer’s uncommitted PENDING row. The third is a consume-transform-produce atomicity misunderstanding: the developer places the Stripe call inside a sendOffsetsToTransaction() loop believing the entire pipeline is atomic, but a JVM crash between producer.send() and commitTransaction() aborts the Kafka transaction without touching the already-committed Stripe charge.

For the tests in this post, the stack is: Micronaut 4.x, micronaut-kafka with kafka.producers.default.transactional-id set, Micronaut Data JPA, PostgreSQL 15 via Testcontainers, Apache Kafka 3.x via Testcontainers (with transactions enabled), WireMock for Stripe API stubs, and JUnit 5 with @MicronautTest.

Background: Kafka EOS — what the guarantee actually covers

Kafka’s exactly-once semantics guarantee is precise: given a transactional producer with a fixed transactional.id, and a consumer with isolation.level=read_committed, each ProducerRecord written inside a transaction that calls commitTransaction() will be visible to the consumer exactly once across all broker failures and producer restarts. Records in transactions that call abortTransaction(), or in transactions that time out before a commit or abort marker is written, are invisible to read_committed consumers.

The guarantee is scoped to the Kafka message delivery pipeline. It says nothing about side effects that happen before, during, or after the Kafka transaction. When the developer writes code that calls Stripe inside a Kafka transaction, the Kafka EOS guarantee provides no coverage for the Stripe call. Stripe’s HTTP response is committed at the HTTP layer the moment Stripe returns HTTP 200. There is no two-phase commit protocol between Kafka and Stripe. There is no mechanism for abortTransaction() to issue an HTTP DELETE to Stripe and cancel a prior successful charge.

This gap between what developers believe EOS guarantees and what it actually guarantees is the root cause of all three failure modes in this post. In each mode, the developer places a Stripe call inside or immediately adjacent to a Kafka transactional operation and reasons that the Kafka atomicity guarantee covers the Stripe call. In each case, it does not.

Mode 1: Kafka producer fencing — abortTransaction() cancels the Kafka message but ch_A stays — new producer epoch consumer replays the record — UUID.randomUUID() generates UUID_B — ch_B committed alongside committed ch_A

The developer builds a payment-processing service that uses a Kafka transactional producer to guarantee exactly-once delivery of payment result events to a downstream topic. The service receives payment commands from an input topic, calls Stripe, and then produces a payment result record to an output topic inside a Kafka transaction. The Stripe call is placed inside the transaction boundary under the reasoning: “If anything goes wrong, I can abort the transaction, and the entire unit of work — including the Stripe charge — will be rolled back.”

// PaymentProducerService.java — unsafe: Stripe call inside Kafka transaction boundary
@Singleton
public class PaymentProducerService {

    private final KafkaProducer<String, PaymentResult> producer;
    private final StripeClient stripeClient;

    public PaymentProducerService(
            @Named("transactional-producer") KafkaProducer<String, PaymentResult> producer,
            StripeClient stripeClient) {
        this.producer = producer;
        this.stripeClient = stripeClient;
    }

    public void processPayment(PaymentCommand command) {
        // UUID generated at call site — new value on every invocation
        String idempotencyKey = UUID.randomUUID().toString();

        try {
            producer.beginTransaction();

            // Step 1: Call Stripe — ch_A committed on Stripe with key_A
            ChargeResult charge = stripeClient.createCharge(
                command.customerId(),
                command.amountCents(),
                idempotencyKey         // key_A on first invocation
            );

            // Step 2: Produce result record to output topic inside transaction
            PaymentResult result = new PaymentResult(
                command.orderId(),
                charge.chargeId(),    // ch_A
                idempotencyKey,
                "COMPLETED"
            );
            producer.send(new ProducerRecord<>("payment-results", command.orderId(), result));

            producer.commitTransaction();
            // Kafka: result record with ch_A is now visible to read_committed consumers

        } catch (Exception e) {
            producer.abortTransaction();
            // Developer belief: "abortTransaction() rolled back everything —
            // the Stripe charge and the Kafka message are both cancelled."
            // Reality: only the Kafka message is cancelled.
            // ch_A is already committed on Stripe.
            throw new RuntimeException("Transaction aborted — retrying", e);
        }
    }
}

The developer’s mental model treats abortTransaction() as a two-phase rollback that covers all operations performed since beginTransaction(). The model is correct for Kafka messages: if abortTransaction() is called, the broker writes an abort marker, and read_committed consumers skip the records in that transaction. The model is wrong for Stripe: the Stripe charge is committed at the HTTP layer the moment Stripe returns HTTP 200 in step 1, independent of whether the Kafka transaction later commits or aborts.

Producer fencing occurs when the Kafka coordinator detects two producer instances with the same transactional.id running concurrently — typically because the application was restarted or scaled horizontally without coordinating transactional.id uniqueness. The Kafka coordinator assigns epochs to transactional producers: each time a producer calls initTransactions(), it receives a new epoch. When a producer with a higher epoch calls initTransactions() for the same transactional.id, the coordinator fences the prior (lower-epoch) producer. Any subsequent call from the fenced producer — including commitTransaction() or abortTransaction() — throws ProducerFencedException. The Kafka runtime, on catching ProducerFencedException, writes an abort marker for any in-flight transaction from the fenced producer epoch.

The sequence of events that produces ch_A and ch_B on the same orderId is:

1. Producer instance 1 (epoch E1) calls producer.beginTransaction().
2. Producer instance 1 generates key_A via UUID.randomUUID().
3. Producer instance 1 calls stripeClient.createCharge(customerId, amountCents, key_A). Stripe processes the charge and returns HTTP 200. ch_A is committed on Stripe.
4. Producer instance 1 calls producer.send(new ProducerRecord<>(...)). The record is written to the broker but not yet visible (transaction open).
5. Producer instance 2 (epoch E2, E2 > E1) calls initTransactions() on the same transactional.id. The Kafka coordinator fences producer E1.
6. Producer instance 1 calls producer.commitTransaction(). The Kafka client throws ProducerFencedException. The broker writes an abort marker for E1’s transaction. The record from step 4 becomes invisible to read_committed consumers.
7. Producer instance 1’s catch block calls producer.abortTransaction(). This throws again (already fenced). The exception propagates to the caller.
8. Producer instance 2 receives the same PaymentCommand (because the command was never acknowledged as processed — the original consumer offset was not committed). Producer instance 2 calls processPayment(command).
9. Producer instance 2 generates key_B via UUID.randomUUID().
10. Producer instance 2 calls stripeClient.createCharge(customerId, amountCents, key_B). Stripe checks its idempotency log: key_B is distinct from key_A (two independent 128-bit random values). Stripe treats this as a new charge request. ch_B is committed on Stripe. The customer’s payment method is debited twice.

Why the developer believes abortTransaction() covers external API calls

Kafka’s EOS documentation describes the transactional API as providing “atomic multi-partition writes.” The word “atomic” is accurate in its Kafka-specific scope: multiple topic-partition writes either all commit or all abort from the perspective of read_committed consumers. Developers familiar with database transactions extrapolate this atomicity to cover all operations inside the beginTransaction() / commitTransaction() block, including HTTP calls. This extrapolation is incorrect. Kafka’s transactional coordinator has no mechanism to communicate with Stripe, no knowledge of HTTP calls, and no rollback protocol for external services.

Fix: content-hash key from the command’s stable fields

The PaymentCommand consumed from the input topic carries the stable fields for the billing event: customerId, orderId, and amountCents. These values are written into the Kafka message payload by the upstream producer and are identical on every replay of the same message. A deterministic key derived from these fields produces the same value on every invocation of processPayment() for the same orderId, across producer epochs and consumer restarts.

// PaymentProducerService.java — safe: content-hash key + Stripe before transaction
@Singleton
public class PaymentProducerService {

    private final KafkaProducer<String, PaymentResult> producer;
    private final StripeClient stripeClient;
    private final PaymentRecordRepository paymentRecordRepository;

    public void processPayment(PaymentCommand command) {
        // Deterministic key — stable across producer epochs and consumer restarts
        String idempotencyKey = Hashing.sha256()
            .hashString(command.customerId() + "|" + command.orderId() + "|" + command.amountCents(),
                        StandardCharsets.UTF_8)
            .toString();

        // Check for prior successful processing before opening the Kafka transaction
        // (handles: consumer restart, producer fencing, message redelivery)
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            // Already processed — idempotent skip
            return;
        }

        // Stripe call BEFORE beginTransaction() — not inside the Kafka atomic boundary
        // If Stripe succeeds and the Kafka transaction later aborts,
        // the PENDING row in the DB ensures the next replay skips the Stripe call.
        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey   // same key on every replay → Stripe returns ch_A, not ch_B
        );

        try {
            producer.beginTransaction();

            // Produce result record inside transaction
            PaymentResult result = new PaymentResult(
                command.orderId(),
                charge.chargeId(),
                idempotencyKey,
                "COMPLETED"
            );
            producer.send(new ProducerRecord<>("payment-results", command.orderId(), result));

            // Save completion record inside the Kafka transaction's database write
            // (requires transactional outbox if strict atomicity is needed)
            paymentRecordRepository.saveCompleted(idempotencyKey, charge.chargeId());

            producer.commitTransaction();

        } catch (Exception e) {
            producer.abortTransaction();
            throw e;
        }
    }
}

With a content-hash key, producer fencing produces the following corrected sequence: producer E2 calls processPayment(command) with the same PaymentCommand as E1. The content-hash key produces key_A (same as E1’s key, because the inputs are identical). Stripe receives the same idempotency key, looks it up in its 24-hour cache, and returns the original ch_A response without creating a new charge. The customer’s payment method is debited exactly once.

Mode 2: read_committed consumer isolation concurrent PENDING guard race — rebalance assigns same partition to two instances — both pass existsByIdempotencyKey() before either PENDING row commits — both call Stripe — ch_B alongside ch_A

The developer has upgraded the consumer to use isolation.level=read_committed, believing this provides strong exactly-once delivery guarantees. The consumer also implements a guard check: before calling Stripe, it calls paymentRecordRepository.existsByIdempotencyKey(idempotencyKey) to skip messages that have already been processed. The developer believes these two measures together make the consumer fully idempotent. The hidden failure is a partition-rebalance race condition that allows two consumer instances to process the same message concurrently, with READ COMMITTED isolation on the database failing to block the second instance’s guard check.

// PaymentConsumer.java — unsafe: guard check before PENDING write,
// rebalance can deliver same record to two instances concurrently
@Singleton
public class PaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    @KafkaListener(
        groupId = "payment-processor",
        isolation = "read_committed"    // only see committed transactional messages
    )
    @Topic("payment-commands")
    @Transactional
    public void onPaymentCommand(PaymentCommand command) {

        String idempotencyKey = buildContentHash(
            command.customerId(),
            command.orderId(),
            String.valueOf(command.amountCents())
        );

        // Guard check — returns true if a prior completed record exists
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            return; // Already processed
        }

        // No record found — proceed to Stripe
        // ⚠ Gap: another consumer instance could pass this check concurrently
        //   if the partition was reassigned during a rebalance before this @Transactional commits

        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey
        );

        // Save COMPLETED record — this @Transactional has not committed yet
        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);
        record.setStripeChargeId(charge.chargeId());
        record.setStatus("COMPLETED");
        paymentRecordRepository.save(record);
        // @Transactional commit happens when this method returns
        // — a rebalance before this commit leaves the offset uncommitted
        //   and the record invisible to other consumers
    }
}

The critical interaction is between Kafka consumer group offset commits and database transaction commits. In Micronaut Kafka with ackMode=AUTO, the consumer group offset is committed after the listener method returns successfully. The database @Transactional commits when the method returns (and the TransactionalInterceptor calls connection.commit()). Between the Stripe call and the method return, there is a window where the charge ch_A has been committed on Stripe but the database row has not been committed and the Kafka offset has not been advanced.

A Kafka consumer group rebalance can occur at any time: due to a new consumer joining the group, a consumer leaving, or a consumer’s heartbeat timeout expiring (for example, because the Stripe call took longer than max.poll.interval.ms). When a rebalance occurs and the current consumer’s partition is revoked, the consumer’s current offset is not committed (the method has not returned). The broker assigns the partition to another consumer instance at the last committed offset, which is the offset before the current message was delivered. The new consumer instance re-reads the same message.

The sequence of events that produces two charges for the same orderId:

1. Consumer instance A receives PaymentCommand for orderId=ORDER-1 at partition offset 42.
2. Consumer instance A calls existsByIdempotencyKey(key_A). The payment_records table has no row with key_A. Guard returns false.
3. Consumer instance A calls stripeClient.createCharge(customerId, amountCents, key_A). Stripe returns HTTP 200. ch_A committed on Stripe.
4. Consumer instance A calls paymentRecordRepository.save(record). The row is inserted inside the open @Transactional. The INSERT is not yet committed. Under READ COMMITTED, this row is invisible to other database connections until the transaction commits.
5. The Stripe call at step 3 takes 4,200 ms, exceeding the consumer’s max.poll.interval.ms of 4,000 ms. The Kafka broker detects the heartbeat timeout and triggers a rebalance. Partition 0 of payment-commands is revoked from consumer instance A and assigned to consumer instance B.
6. Consumer instance B starts polling partition 0 at the last committed offset, which is offset 41 (the offset before the PaymentCommand that instance A is processing). Consumer instance B reads offset 42: the same PaymentCommand for orderId=ORDER-1.
7. Consumer instance B calls existsByIdempotencyKey(key_A). This is a separate database transaction on a different connection. Under READ COMMITTED, this connection sees only committed rows. Consumer instance A’s INSERT at step 4 is still uncommitted. The guard returns false.
8. Consumer instance B generates key_B via UUID.randomUUID() — wait, no: in this scenario the content-hash key is used, so key_B = key_A. But wait — if key_B = key_A and Stripe has key_A cached, Stripe would return ch_A. The real failure here requires that instance B also uses a random UUID. Let me reconsider.

The race condition is most damaging when the developer uses a content-hash key for Stripe but the guard check alone (without a PENDING row first) allows concurrent calls. If consumer instance B also uses the content-hash key_A (derived from the stable message fields), then Stripe’s idempotency cache returns ch_A for instance B’s call — the charge is deduplicated at Stripe. However, this deduplication only holds within Stripe’s 24-hour idempotency window. The real failure occurs when the developer uses UUID.randomUUID() at method entry: instance A generates key_A, instance B generates key_B. Stripe sees two distinct keys, creates ch_A and ch_B.

With UUID.randomUUID(), consumer instance B’s sequence continues:

8. Consumer instance B generates key_B via UUID.randomUUID().
9. Consumer instance B calls stripeClient.createCharge(customerId, amountCents, key_B). Stripe checks its idempotency log: key_B is not in the log (key_A is, from instance A’s call, but key_B is a distinct UUID). Stripe creates a new charge. ch_B committed on Stripe.
10. Consumer instance A’s @Transactional eventually commits (or rolls back due to the rebalance). If it commits, the database has a row with key_A and ch_A but no application-level awareness of ch_B.
11. Consumer instance B’s @Transactional commits. The database has a row with key_B and ch_B. The customer’s payment method has been debited twice.

Why isolation.level=read_committed does not prevent concurrent consumer processing

isolation.level=read_committed on a Kafka consumer controls which Kafka messages are visible to the consumer: only messages from committed transactional producers, filtering out aborted and in-flight transactions. It is a Kafka-layer setting. It says nothing about whether two consumer instances can process the same Kafka message concurrently. A Kafka consumer group partitions the topic among consumer instances. Normally, each partition is assigned to exactly one consumer instance at a time — but “at a time” is defined by the Kafka rebalance protocol, which has a gap between offset commit and partition revocation where the same message can be delivered to two instances.

Fix: PENDING row INSERT before Stripe with UNIQUE constraint as concurrent guard

The correct pattern is to insert a PENDING row with a UNIQUE constraint on idempotencyKey before calling Stripe, inside a transaction that commits the row before the Stripe call. This serializes concurrent consumer instances at the database level: only one will succeed in inserting the PENDING row. The other will get a DataIntegrityViolationException on the INSERT and must not proceed to Stripe.

// PaymentConsumer.java — safe: PENDING row before Stripe + content-hash key
@Singleton
public class PaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    @KafkaListener(
        groupId = "payment-processor",
        isolation = "read_committed"
    )
    @Topic("payment-commands")
    public void onPaymentCommand(PaymentCommand command) {

        String idempotencyKey = buildContentHash(
            command.customerId(),
            command.orderId(),
            String.valueOf(command.amountCents())
        );

        // Check for prior COMPLETED processing
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            return; // Already processed — idempotent skip
        }

        // Insert PENDING row in its own short-lived transaction
        // that commits BEFORE the Stripe call.
        // This serializes concurrent consumer instances:
        // the second one gets DataIntegrityViolationException on the UNIQUE key.
        try {
            paymentRecordRepository.insertPending(idempotencyKey, command);
            // This transaction commits here
        } catch (DataIntegrityViolationException e) {
            // Another instance inserted the PENDING row first.
            // The Kafka offset for this instance was not committed (we're not returning normally),
            // so Kafka will redeliver the message. By then, the winning instance will have
            // committed the COMPLETED row and the existsByIdempotencyKey() check will return true.
            return;
        }

        // Only one consumer instance reaches here for a given idempotencyKey.
        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey  // content-hash key — Stripe deduplicates on replay
        );

        // Update PENDING → COMPLETED in a separate transaction
        paymentRecordRepository.markCompleted(idempotencyKey, charge.chargeId());
    }
}

The insertPending() method runs in its own @Transactional(propagation=REQUIRES_NEW): it opens a separate JDBC connection, inserts the PENDING row, and commits. The commit makes the row visible to all other database connections under any isolation level, including READ COMMITTED. When consumer instance B checks existsByIdempotencyKey() after instance A’s PENDING row is committed, the check returns true and instance B exits without calling Stripe. If instance B checks before instance A commits, instance B will attempt to insert its own PENDING row and receive DataIntegrityViolationException on the UNIQUE constraint, preventing it from reaching the Stripe call.

Mode 3: sendOffsetsToTransaction() consume-transform-produce loop — Stripe call outside the Kafka atomicity boundary — JVM crash after producer.send() aborts the Kafka transaction — consumer re-reads the record — UUID.randomUUID() generates UUID_B — ch_B committed alongside ch_A

The developer implements an EOS consume-transform-produce loop: consume a payment command from an input topic, call Stripe, transform the result, produce a payment result to an output topic, and atomically bind the consumer offset advance and the produced message to a single Kafka transaction via sendOffsetsToTransaction(). The developer’s intent is to make the entire pipeline — consume, process, produce — atomic: either the input offset advances and the output message is committed, or neither happens. The Stripe call is placed inside the loop on the assumption that it is part of the “transform” step and is therefore covered by the atomicity guarantee.

// EosPaymentLoop.java — unsafe: Stripe call inside consume-transform-produce loop
@Singleton
public class EosPaymentLoop {

    private final KafkaConsumer<String, PaymentCommand> consumer;
    private final KafkaProducer<String, PaymentResult> producer;
    private final StripeClient stripeClient;
    private final TopicPartition inputPartition = new TopicPartition("payment-commands", 0);

    public void run() {
        consumer.subscribe(List.of("payment-commands"));
        producer.initTransactions(); // initialize EOS; receives producer epoch

        while (!Thread.currentThread().isInterrupted()) {
            ConsumerRecords<String, PaymentCommand> records = consumer.poll(Duration.ofMillis(500));

            for (ConsumerRecord<String, PaymentCommand> record : records) {
                PaymentCommand command = record.value();

                // UUID generated per record — new value on every invocation
                String idempotencyKey = UUID.randomUUID().toString();

                try {
                    producer.beginTransaction();

                    // Step 1: Call Stripe — ch_A committed on Stripe
                    // Developer believes: "I'm inside a Kafka transaction —
                    // abortTransaction() will undo this Stripe charge."
                    ChargeResult charge = stripeClient.createCharge(
                        command.customerId(),
                        command.amountCents(),
                        idempotencyKey  // key_A — random, not stable
                    );

                    // Step 2: Produce result to output topic inside the transaction
                    PaymentResult result = new PaymentResult(
                        command.orderId(), charge.chargeId(), "COMPLETED"
                    );
                    producer.send(new ProducerRecord<>("payment-results", command.orderId(), result));

                    // Step 3: Atomically bind consumer offset advance to this transaction
                    // so that "offset advanced" and "result message visible" are atomic
                    Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(
                        inputPartition,
                        new OffsetAndMetadata(record.offset() + 1)
                    );
                    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

                    // Step 4: Commit — makes result visible to read_committed consumers
                    //          and advances the consumer offset atomically
                    producer.commitTransaction();
                    // ⚠ JVM crash here (OOM, SIGKILL, network partition)
                    //   Kafka coordinator: transaction timeout → write abort marker
                    //   result on output topic: INVISIBLE to read_committed consumers
                    //   consumer offset: ROLLED BACK — stays at record.offset()
                    //   Stripe charge ch_A: COMMITTED — not rolled back

                } catch (Exception e) {
                    producer.abortTransaction();
                    // abortTransaction() cancels the Kafka messages and the offset advance.
                    // It does not cancel the Stripe charge ch_A.
                    throw e;
                }
            }
        }
    }
}

The sendOffsetsToTransaction() API is one of Kafka’s most powerful EOS primitives. It takes a map of TopicPartition → OffsetAndMetadata and the consumer group’s metadata (group ID + generation ID) and writes the offset commit as a Kafka message to the __consumer_offsets internal topic, inside the current Kafka transaction. When commitTransaction() is called, both the produced records on the output topic and the consumer offset advance on the input topic are committed atomically. When abortTransaction() is called, both the produced records and the offset advance are rolled back: the output messages are invisible to read_committed consumers, and the consumer offset stays at its previous committed value — the consumer re-reads the input record on the next poll.

This atomicity is precisely what the developer wants for the Kafka pipeline. The mistake is placing the Stripe call at step 1, before producer.send() and sendOffsetsToTransaction(). The Stripe call is not a Kafka operation. It is an HTTP request to Stripe’s API server over the internet. It is not included in the Kafka transaction’s commit or abort protocol. When abortTransaction() is called — whether explicitly (in the catch block) or implicitly (by the Kafka coordinator after a transaction timeout due to JVM crash) — Stripe is not notified. ch_A remains committed on Stripe’s side.

The step-by-step failure sequence with a JVM crash between producer.send() and commitTransaction():

1. The EOS loop reads PaymentCommand for orderId=ORDER-1 at offset 42.
2. producer.beginTransaction() opens a Kafka transaction (epoch E1, transaction sequence T1).
3. UUID.randomUUID() generates key_A.
4. stripeClient.createCharge(customerId, amountCents, key_A) — Stripe returns HTTP 200. ch_A committed on Stripe.
5. producer.send(new ProducerRecord<>("payment-results", ...)) — result record is written to the broker inside transaction T1. Not yet visible to read_committed consumers.
6. JVM receives SIGKILL (OOM killer, deployment rollout, hardware failure). The process terminates before sendOffsetsToTransaction() and commitTransaction() are called.
7. The Kafka coordinator detects that producer E1 has not sent a transaction heartbeat for transaction.timeout.ms (default 60 seconds). The coordinator writes an abort marker for transaction T1 to the payment-results partition log. The result record from step 5 becomes invisible to read_committed consumers.
8. The consumer offset for payment-commands partition 0 was not advanced (step 3’s sendOffsetsToTransaction() was never called). The committed offset remains at 41.
9. The JVM restarts. The EOS loop re-initializes: consumer.subscribe(), producer.initTransactions() (receives new epoch E2). consumer.poll() returns offset 42: the same PaymentCommand for orderId=ORDER-1.
10. producer.beginTransaction() opens transaction T2 (epoch E2).
11. UUID.randomUUID() generates key_B. (A new random UUID; the prior key_A is not stored anywhere accessible to the restarted process.)
12. stripeClient.createCharge(customerId, amountCents, key_B). Stripe checks its idempotency log: key_A is in the log (from step 4, within the 24-hour window), but key_B is distinct. Stripe creates a new charge. ch_B committed on Stripe. The customer’s payment method is debited twice.

Why the developer believes sendOffsetsToTransaction() covers the Stripe call

The consume-transform-produce EOS pattern is often described as “read-process-write atomicity”: the act of consuming a message, transforming it, and producing the result are atomic. This description is accurate when “process” means a pure, side-effect-free transformation — a map operation, a filter, an aggregation. When “process” includes an external API call with its own commit semantics (Stripe, Twilio, Resend, a payment gateway), the atomicity guarantee no longer covers the processing step. The developer conflates “atomic read-process-write in the Kafka pipeline” with “atomic execution of all operations inside the loop.” These are not the same thing.

Fix: content-hash key + existence check before Stripe + Stripe outside the Kafka transaction boundary

// EosPaymentLoop.java — safe: content-hash key + check + Stripe before beginTransaction()
@Singleton
public class EosPaymentLoop {

    private final KafkaConsumer<String, PaymentCommand> consumer;
    private final KafkaProducer<String, PaymentResult> producer;
    private final StripeClient stripeClient;
    private final PaymentRecordRepository paymentRecordRepository;

    public void run() {
        consumer.subscribe(List.of("payment-commands"));
        producer.initTransactions();

        while (!Thread.currentThread().isInterrupted()) {
            ConsumerRecords<String, PaymentCommand> records = consumer.poll(Duration.ofMillis(500));

            for (ConsumerRecord<String, PaymentCommand> record : records) {
                PaymentCommand command = record.value();

                // Deterministic key from stable message fields
                String idempotencyKey = buildContentHash(
                    command.customerId(), command.orderId(),
                    String.valueOf(command.amountCents())
                );

                // Skip if already processed (handles: restart, rebalance, replay)
                Optional<PaymentRecord> existing =
                    paymentRecordRepository.findByIdempotencyKey(idempotencyKey);
                if (existing.isPresent()) {
                    // Already processed — advance offset only
                    advanceOffset(record);
                    continue;
                }

                // Stripe call OUTSIDE the Kafka transaction.
                // If this JVM crashes after the Stripe call and before commitTransaction(),
                // the consumer re-reads the record and the content-hash key hits Stripe's
                // idempotency cache — Stripe returns ch_A, not ch_B.
                ChargeResult charge = stripeClient.createCharge(
                    command.customerId(), command.amountCents(), idempotencyKey
                );

                try {
                    producer.beginTransaction();

                    // Produce result inside transaction
                    PaymentResult result = new PaymentResult(
                        command.orderId(), charge.chargeId(), idempotencyKey, "COMPLETED"
                    );
                    producer.send(new ProducerRecord<>(
                        "payment-results", command.orderId(), result
                    ));

                    // Advance consumer offset atomically with the result message
                    Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(
                        new TopicPartition("payment-commands", record.partition()),
                        new OffsetAndMetadata(record.offset() + 1)
                    );
                    producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

                    // Save COMPLETED record inside transaction (transactional outbox approach)
                    // — note: this requires a JTA-capable transaction manager bridging Kafka
                    // and JDBC, which is complex. The simpler approach: save before beginTransaction()
                    // and accept that the DB record may exist without a committed Kafka message.
                    producer.commitTransaction();

                } catch (Exception e) {
                    producer.abortTransaction();
                    // On retry (Kafka re-reads at last committed offset):
                    // - idempotencyKey is the same (content-hash from stable fields)
                    // - Stripe returns ch_A from its idempotency cache (if within 24h)
                    // - no double charge
                    throw e;
                }
            }
        }
    }
}

Comparison table

Integration tests

Each mode can be reproduced deterministically in a @MicronautTest with Testcontainers (Kafka 3.x, PostgreSQL 15) and WireMock. The key setup for each:

Mode 1 (producer fencing) test setup

@MicronautTest
class ProducerFencingTest {

    @Inject
    EosPaymentLoop loop;

    @Inject
    PaymentRecordRepository repository;

    WireMockServer wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(8089));

    @BeforeEach
    void setup() {
        wireMock.start();
        // Stripe stub: always succeeds — records every call
        wireMock.stubFor(post(urlPathEqualTo("/v1/charges"))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("""
                    {"id":"ch_test_001","object":"charge","amount":2000,"status":"succeeded"}
                    """)));
    }

    @Test
    void fencedProducerReplaysMessage_contentHashKeyDeduplicates() throws Exception {
        // Given: a PaymentCommand on the input topic
        PaymentCommand command = new PaymentCommand("cust_1", "order_1", 2000);
        publishToKafka("payment-commands", command);

        // When: producer instance 1 processes and gets fenced mid-transaction
        // (simulated by forcing a second initTransactions() call with same transactional.id)
        simulateFencingDuringProcessing();

        // When: producer instance 2 replays the same message
        loop.run();

        // Then: Stripe was called twice (once per producer epoch), but ch_A was returned
        // on the second call because the content-hash key matches
        wireMock.verify(2, postRequestedFor(urlPathEqualTo("/v1/charges")));

        // Then: only one payment record in the database
        List<PaymentRecord> records = repository.findByOrderId("order_1");
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getStripeChargeId()).isEqualTo("ch_test_001");
    }

    @Test
    void fencedProducerWithRandomUUID_createsDoubleCharge() throws Exception {
        // Given: a PaymentCommand on the input topic
        PaymentCommand command = new PaymentCommand("cust_1", "order_1", 2000);
        publishToKafka("payment-commands", command);

        // When: producer instance 1 processes, calls Stripe (ch_A), gets fenced
        // producer instance 2 replays, calls Stripe with new UUID (ch_B)
        runTwoEpochsWithRandomUUID();

        // Then: Stripe was called twice with distinct idempotency keys
        wireMock.verify(2, postRequestedFor(urlPathEqualTo("/v1/charges")));

        // Then: two payment records — double charge
        List<PaymentRecord> records = repository.findByOrderId("order_1");
        assertThat(records).hasSize(2); // BUG: two charges for the same order
    }
}

Mode 2 (concurrent guard race) test setup

@MicronautTest
class ConcurrentConsumerGuardRaceTest {

    @Inject
    ApplicationContext ctx;

    @Test
    void twoConsumerInstances_pendingRowGuard_preventsDoubleCharge() throws Exception {
        WireMockServer stripe = new WireMockServer(8089);
        stripe.start();
        AtomicInteger stripeCallCount = new AtomicInteger(0);
        stripe.stubFor(post(urlPathEqualTo("/v1/charges"))
            .willReturn(aResponse()
                .withStatus(200)
                .withBody("""
                    {"id":"ch_test_pending","object":"charge","status":"succeeded"}
                    """)));

        PaymentCommand command = new PaymentCommand("cust_1", "order_1", 5000);

        // Simulate two consumer instances processing the same record concurrently
        // using a CountDownLatch to synchronize both past the guard check
        // before either inserts a PENDING row
        CountDownLatch guardPassed = new CountDownLatch(2);
        CountDownLatch releaseStripe = new CountDownLatch(1);

        Runnable consumer1Task = () -> processWithLatch(command, guardPassed, releaseStripe);
        Runnable consumer2Task = () -> processWithLatch(command, guardPassed, releaseStripe);

        var f1 = CompletableFuture.runAsync(consumer1Task);
        var f2 = CompletableFuture.runAsync(consumer2Task);
        guardPassed.await(5, TimeUnit.SECONDS);
        releaseStripe.countDown(); // release both to Stripe simultaneously

        f1.join();
        f2.join();

        // With PENDING row + UNIQUE constraint: only one Stripe call succeeds past the guard
        stripe.verify(1, postRequestedFor(urlPathEqualTo("/v1/charges")));
    }
}

Mode 3 (sendOffsetsToTransaction() crash) test setup

@MicronautTest
class SendOffsetsTransactionCrashTest {

    @Test
    void jvmCrashBetweenSendAndCommit_contentHashKeyPreventsDoubleCharge() throws Exception {
        WireMockServer stripe = new WireMockServer(8089);
        stripe.start();
        stripe.stubFor(post(urlPathEqualTo("/v1/charges"))
            .willReturn(aResponse().withStatus(200)
                .withBody("{\"id\":\"ch_eos_001\",\"status\":\"succeeded\"}")));

        PaymentCommand command = new PaymentCommand("cust_1", "order_1", 3000);
        publishToKafka("payment-commands", command);

        // First run: Stripe call succeeds, JVM "crashes" (simulated) before commitTransaction()
        runLoopWithCrashAfterSend();
        // At this point: Kafka transaction aborted, consumer offset not advanced,
        // Stripe ch_eos_001 committed.

        // Second run: same record replayed with content-hash key
        EosPaymentLoop loop = ctx.getBean(EosPaymentLoop.class);
        loop.run();

        // Stripe called twice — but second call returns ch_eos_001 from idempotency cache
        // (content-hash key is the same for both invocations)
        stripe.verify(2, postRequestedFor(urlPathEqualTo("/v1/charges")));

        // Assert: Stripe idempotency log contains only key_A, not a distinct key_B
        // (verified by checking the idempotency key header on both requests)
        List<LoggedRequest> stripeRequests = stripe.findAll(
            postRequestedFor(urlPathEqualTo("/v1/charges")));
        String key1 = extractIdempotencyKey(stripeRequests.get(0));
        String key2 = extractIdempotencyKey(stripeRequests.get(1));
        assertThat(key1).isEqualTo(key2); // same content-hash key on both invocations

        // Assert: only one payment record in the database
        assertThat(paymentRecordRepository.findByOrderId("order_1")).hasSize(1);
    }
}

Summary

All three failure modes in this post share a single root cause: the developer believes Kafka EOS atomicity covers external API calls made inside the Kafka transactional boundary. It does not. Kafka’s transactional atomicity is scoped to Kafka topic-partition writes and consumer offset commits. Stripe’s commit happens at the HTTP layer, independently of the Kafka transaction lifecycle. No form of Kafka EOS — not producer fencing, not read_committed isolation, not sendOffsetsToTransaction() — provides a rollback mechanism for committed Stripe charges.

The fix in all three modes is the same two-part pattern: (1) derive the Stripe idempotency key from the stable fields of the Kafka message (a content-hash of customerId + orderId + amountCents), not from UUID.randomUUID(); and (2) check for prior processing before calling Stripe, using a PENDING row with a UNIQUE constraint to serialize concurrent consumer instances. The content-hash key ensures that any replay of the same Kafka message — across producer epochs, JVM restarts, and partition rebalances — sends the same Stripe idempotency key, allowing Stripe’s 24-hour cache to return the original charge without creating a new one. The PENDING row ensures that concurrent consumer instances cannot both pass the guard check and both reach Stripe simultaneously.

These patterns apply regardless of whether the Micronaut application uses Micronaut Kafka’s high-level @KafkaListener / @KafkaClient annotations (covered in the prior post) or the lower-level EOS APIs covered here. The Kafka layer determines how and when messages are delivered. The Stripe idempotency layer determines whether a message delivery results in one charge or two. A correct Stripe integration must be designed to survive all message delivery scenarios the Kafka layer can produce — including producer fencing, partition rebalance, and consume-transform-produce abort.