Spring Kafka Exactly-Once Semantics and Stripe Integration: How DefaultErrorHandler Re-delivers the ConsumerRecord to @KafkaListener from Line One Generating UUID_B While ch_A Stays Committed on Stripe, @Retryable on a @Transactional(kafkaTransactionManager) Service Bean Calls MethodInvocation.proceed() on Retry Re-entering the Method Body and Regenerating UUID at the Top, and ChainedKafkaTransactionManager Sequential Commit Permanently Commits JDBC Before Kafka Fails — Consumer Re-reads with UUID_B

Three Spring Kafka EOS Stripe idempotency failure modes, each mechanically distinct from all prior Spring Kafka and Micronaut Kafka posts in this series. This post covers Spring Kafka’s runtime classes that developers commonly combine with Stripe billing: the container-level DefaultErrorHandler, the AOP layering of @Retryable over @Transactional(kafkaTransactionManager), and the ChainedKafkaTransactionManager that was designed to coordinate Kafka and JDBC transactions atomically but does not provide two-phase commit. In all three modes the developer holds a coherent mental model that predicts safety — and in all three modes the model is wrong in a way that is invisible in unit tests and only manifests as a double charge under specific runtime conditions.

The Micronaut Kafka EOS post covered three Micronaut-specific EOS failure modes: transactional producer fencing leaving ch_A committed after abortTransaction(), read_committed isolation concurrent PENDING guard race, and sendOffsetsToTransaction() atomicity not covering the Stripe HTTP call. This post covers three failure modes that are specific to Spring Kafka’s runtime infrastructure and do not map directly to any of those three. The failure sequence in all three Spring Kafka modes ends the same way: the listener method or service method is re-entered from line one while ch_A is already committed on Stripe, UUID.randomUUID() generates UUID_B, and Stripe creates ch_B.

Mode 1: DefaultErrorHandler seek-and-re-poll re-delivers ConsumerRecord to @KafkaListener from line 1 — UUID.randomUUID() at method entry generates UUID_B

Spring Kafka’s DefaultErrorHandler (the default error handler since Spring Kafka 2.8, replacing SeekToCurrentErrorHandler) is a container-level recovery handler. When the @KafkaListener method throws an exception, the container calls DefaultErrorHandler.handleAndRecover(). The handler calls consumer.seek(TopicPartition, offset) to reset the consumer partition position back to the offset of the failed record. After the configured BackOff interval, the container re-polls the broker. Because the partition was seeked back, the broker returns the same ConsumerRecord from the original offset. The container delivers it to the @KafkaListener method as a fresh call from line 1.

The following listener places UUID.randomUUID() at method entry, calls Stripe before writing to the database, and relies on DefaultErrorHandler to retry on DataAccessException:

@Component
class PaymentListener {

    @Autowired StripeClient stripeClient;
    @Autowired PaymentRecordRepository paymentRecordRepository;

    @KafkaListener(topics = "payment-commands", groupId = "billing-group")
    void handlePayment(ConsumerRecord<String, PaymentCommand> record) {
        PaymentCommand command = record.value();
        String idempotencyKey = UUID.randomUUID().toString(); // UUID_A on attempt 1

        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(),
            command.getCustomerId(),
            idempotencyKey  // Stripe POST /v1/charges with Idempotency-Key: UUID_A
        );

        // Throws DataAccessException (e.g., unique constraint on order_id,
        // connection pool exhaustion, or a transient deadlock)
        paymentRecordRepository.save(
            new PaymentRecord(command.getOrderId(), idempotencyKey, charge.getId(), "COMPLETED")
        );
    }
}

// DefaultErrorHandler configured explicitly with FixedBackOff(1000ms, 3 retries)
@Bean
DefaultErrorHandler errorHandler() {
    return new DefaultErrorHandler(new FixedBackOff(1000L, 3));
}

Step-by-step failure trace

1. Container polls payment-commands; broker returns PaymentCommand(customerId=“cust_1”, orderId=“order_1”, amountCents=5000) at partition 0, offset 42.
2. Container calls handlePayment(record). UUID.randomUUID() at line 1 of the method body executes. 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. paymentRecordRepository.save() throws DataAccessException. The exception propagates out of handlePayment().
5. Spring Kafka’s container catches the exception and calls DefaultErrorHandler.handleAndRecover(data, thrownException). data is the list of failed ConsumerRecord objects (one in this case: the record at partition 0, offset 42).
6. DefaultErrorHandler calls consumer.seek(new TopicPartition(“payment-commands”, 0), 42). The consumer’s position for partition 0 is now back at offset 42. No Kafka offset commit has been made — the consumer’s last committed offset for this partition is still wherever it was before offset 42 was fetched.
7. The container applies the FixedBackOff sleep (1000ms). After the interval the container calls consumer.poll(Duration.ofMillis(3000)). Because the partition was seeked to offset 42, the broker returns the same ConsumerRecord from offset 42.
8. Container calls handlePayment(record) again, with the same record from offset 42. The method is invoked from line 1.
9. UUID.randomUUID() at line 1 executes again. idempotencyKey = UUID_B. This is a new UUID.randomUUID() call, producing a new random value distinct from UUID_A.
10. 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 now committed on Stripe alongside ch_A.

Developer misconception

“Spring Kafka’s DefaultErrorHandler is a recovery mechanism that targets the failed operation. Since paymentRecordRepository.save() was the failing line, the handler will retry that specific step. The Stripe call completed successfully on attempt 1 — the error handler will not re-execute the Stripe call.”

This model assumes DefaultErrorHandler has visibility into the internal execution of the listener method — that it knows which lines succeeded and which line threw, and re-enters the method at the line after the failure. That is not how it works. DefaultErrorHandler operates at the Kafka consumer offset level. It knows two things: the ConsumerRecord that was being processed and the exception that was thrown. It has no knowledge of the method body, its internal branches, or what side effects have occurred. Its only action is consumer.seek(partition, offset) followed by a re-poll. The consequence is that the entire @KafkaListener method is called again from line 1 — including every line before the one that threw. UUID.randomUUID() is one of those lines.

The same pattern produces the same result regardless of where in the method body the exception is thrown. If DataAccessException is thrown on line 20 and the Stripe call is on line 5, the Stripe call executes again on every DefaultErrorHandler re-delivery. The listener method is always re-called from line 1.

Fix

Two changes are needed. First, replace UUID.randomUUID() with a content-hash key derived from the stable fields of the PaymentCommand. A content-hash key is identical on every re-delivery of the same logical billing event — the same customerId, orderId, and amountCents always produce the same key. Stripe’s 24h idempotency cache returns ch_A when it receives the same key on a re-delivery, without creating a new charge. Second, insert a PENDING row before the Stripe call (in a REQUIRES_NEW transaction so it commits immediately) and check for the row before calling Stripe. The existence check + PENDING row combination handles both the re-delivery case and concurrent consumers processing the same partition in a rebalance window.

@KafkaListener(topics = "payment-commands", groupId = "billing-group")
void handlePayment(ConsumerRecord<String, PaymentCommand> record) {
    PaymentCommand command = record.value();

    // Content-hash key: stable across any number of DefaultErrorHandler re-deliveries
    // of the same PaymentCommand. SHA-256 of customerId + orderId + amountCents.
    String idempotencyKey = contentHashKey(command);

    // Guard: PENDING or COMPLETED row already exists — skip Stripe
    Optional<PaymentRecord> existing =
        paymentRecordRepository.findByIdempotencyKey(idempotencyKey);
    if (existing.isPresent()) {
        // Already processed or in-progress on a concurrent consumer instance.
        // Kafka will commit the consumer offset for this record on normal return.
        return;
    }

    // PENDING INSERT in REQUIRES_NEW so it commits immediately and is visible
    // to any concurrent consumer that re-reads this partition offset.
    savePending(command, idempotencyKey);

    // Stripe — content-hash key means DefaultErrorHandler re-delivery #1, #2, #3
    // all send the same Idempotency-Key header. Stripe returns ch_A from cache.
    StripeCharge charge = stripeClient.createCharge(
        command.getAmountCents(), command.getCustomerId(), idempotencyKey);

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

@Transactional(propagation = Propagation.REQUIRES_NEW)
void savePending(PaymentCommand command, String idempotencyKey) {
    paymentRecordRepository.save(
        new PaymentRecord(command.getOrderId(), idempotencyKey, null, "PENDING")
    );
}

private 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, all three DefaultErrorHandler re-deliveries of the same PaymentCommand arrive at the Stripe API with the same Idempotency-Key header. The first delivery creates ch_A. Deliveries two and three find the key in Stripe’s cache and return ch_A without creating a new charge. After exhausting retries, if DefaultErrorHandler calls the optional recovery callback, that callback must also check for the COMPLETED row before any further Stripe interaction.

Mode 2: @Retryable on a @Transactional(kafkaTransactionManager) service bean — RetryOperationsInterceptor is outer, TransactionInterceptor is inner — MethodInvocation.proceed() on retry re-opens the Kafka transaction and re-runs the method body from line 1

A common Spring Kafka pattern: a service bean method is annotated with both @Transactional(transactionManager = “kafkaTransactionManager”) (to make outbound kafkaTemplate.send() calls part of a Kafka transaction) and @Retryable (to retry transient DataAccessExceptions). The @KafkaListener method calls this service bean. The failure arises from the AOP interceptor ordering Spring uses by default:

• RetryOperationsInterceptor from @Retryable: default order Integer.MAX_VALUE - 5. Lower order number = higher priority = outer position in the AOP proxy.
• TransactionInterceptor from @Transactional(kafkaTransactionManager): default order Integer.MAX_VALUE. Higher order number = lower priority = inner position in the AOP proxy.

The AOP proxy chain for the service method is:

Caller (@KafkaListener method)
  └── RetryOperationsInterceptor          [order = Integer.MAX_VALUE - 5, OUTER]
        └── TransactionInterceptor        [order = Integer.MAX_VALUE, INNER]
              └── BillingService.processBilling() method body

The buggy implementation:

// @KafkaListener method calls the service bean
@KafkaListener(topics = "payment-commands", groupId = "billing-group")
void handlePayment(ConsumerRecord<String, PaymentCommand> record) {
    billingService.processBilling(record.value());
}

@Service
class BillingService {

    @Autowired StripeClient stripeClient;
    @Autowired KafkaTemplate<String, PaymentResult> kafkaTemplate;
    @Autowired PaymentRecordRepository paymentRecordRepository;

    // @Retryable is OUTER in the AOP proxy (order Integer.MAX_VALUE - 5)
    // @Transactional(kafkaTxManager) is INNER (order Integer.MAX_VALUE)
    @Retryable(
        retryFor = {DataAccessException.class},
        maxAttempts = 3,
        backoff = @Backoff(delay = 500)
    )
    @Transactional(transactionManager = "kafkaTransactionManager")
    public void processBilling(PaymentCommand command) {
        String idempotencyKey = UUID.randomUUID().toString(); // UUID_A on attempt 1

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

        // Outbound Kafka message — inside KafkaTransactionManager transaction
        kafkaTemplate.send("payment-results", command.getOrderId(),
            new PaymentResult(command.getOrderId(), charge.getId(), idempotencyKey));

        // DB write — throws DataAccessException on attempt 1
        paymentRecordRepository.save(
            new PaymentRecord(command.getOrderId(), idempotencyKey, charge.getId()));
    }
}

Step-by-step failure trace

Attempt 1:

1. RetryOperationsInterceptor.invoke() is entered. It calls MethodInvocation.proceed() to execute the next interceptor in the chain.
2. TransactionInterceptor opens a new KafkaTransactionManager transaction. The DefaultKafkaProducerFactory assigns a transactional producer with transactionId = kafkaTxPrefix.0; the producer calls KafkaProducer.beginTransaction().
3. Method body executes. UUID.randomUUID() → idempotencyKey = UUID_A.
4. stripeClient.createCharge(5000, “cust_1”, UUID_A) → ch_A committed on Stripe (HTTP 200).
5. kafkaTemplate.send(“payment-results”, …) → message buffered inside the open Kafka transaction (not yet visible to read_committed consumers).
6. paymentRecordRepository.save() throws DataAccessException. Exception propagates to TransactionInterceptor.
7. TransactionInterceptor calls kafkaTransactionManager.rollback(). The Kafka producer calls KafkaProducer.abortTransaction(): the buffered message on payment-results is aborted; an abort marker is written to the partition log; no read_committed consumer will see the message. The Stripe charge ch_A is not affected — Stripe has no JTA enrollment and cannot be rolled back via abortTransaction().
8. Exception propagates to RetryOperationsInterceptor. DataAccessException matches the retryFor predicate. Retry is scheduled after 500ms backoff.

Attempt 2:

1. RetryOperationsInterceptor calls MethodInvocation.proceed() again on the SAME MethodInvocation object. This call traverses the entire AOP interceptor chain starting from the next interceptor — TransactionInterceptor.
2. TransactionInterceptor opens a NEW KafkaTransactionManager transaction. The transactional producer begins a new transaction.
3. Method body re-executes from line 1. UUID.randomUUID() → idempotencyKey = UUID_B. This is a fresh call to randomUUID() producing a new value distinct from UUID_A.
4. stripeClient.createCharge(5000, “cust_1”, UUID_B) → Stripe POST /v1/charges with Idempotency-Key: UUID_B. Stripe looks up UUID_B: it is not in the 24h cache (UUID_A is, but UUID_B is a different key). Stripe creates ch_B alongside ch_A.

Developer misconception

“@Retryable is configured to retry DataAccessException, which is a JDBC-layer exception. The @Transactional(kafkaTransactionManager) annotation wraps the Kafka producer transaction for the kafkaTemplate.send() call inside the method. The Stripe call at the top of the method is not a JDBC operation and is not inside the KafkaTransactionManager transaction boundary. @Retryable will re-execute only the JDBC part of the operation, not the Stripe call.”

This model treats @Retryable as a selective retry that can somehow skip lines that already succeeded. That is not how AOP interceptors work. Both @Retryable and @Transactional are Spring AOP interceptors wrapping the same method invocation point. When RetryOperationsInterceptor calls MethodInvocation.proceed() on a retry, it re-invokes the AOP interceptor chain starting at TransactionInterceptor, which in turn invokes the method body in its entirety from line 1. There is no mechanism for any AOP interceptor to skip specific lines of the method body based on which ones executed successfully in a prior attempt. The @Transactional(kafkaTransactionManager) boundary does not protect the Stripe call — the Stripe call is inside the method body and is re-executed on every retry.

This failure mode differs from the Micronaut Kafka posts in an important way: here the retry is on a service bean called by the @KafkaListener method, not on the @KafkaListener method itself. The @KafkaListener method succeeds (no exception escapes it) as long as @Retryable exhausts its attempts on the service bean and finally throws. The Kafka consumer offset IS committed after the @KafkaListener completes — which means there is no Kafka-level re-delivery of the original message. The double charge is caused entirely by @Retryable’s multiple calls to MethodInvocation.proceed() on the service bean, not by the Kafka transport layer replaying a record.

Fix

The content-hash key is the primary fix. On attempt 2, processBilling() is called with the same PaymentCommand (same customerId, orderId, amountCents) — the content-hash produces the same key as attempt 1. Stripe returns ch_A from its 24h cache. There is no ch_B.

@Service
class BillingService {

    @Retryable(retryFor = {DataAccessException.class}, maxAttempts = 3,
               backoff = @Backoff(delay = 500))
    @Transactional(transactionManager = "kafkaTransactionManager")
    public void processBilling(PaymentCommand command) {
        // Content-hash: same value on every @Retryable attempt for the same command
        String idempotencyKey = contentHashKey(command);

        // Guard: PENDING or COMPLETED row already exists (from a prior @Retryable attempt)
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            // Prior attempt committed Stripe charge ch_A. Skip Stripe.
            // If the prior attempt also committed the Kafka message, skip that too.
            return;
        }

        // PENDING row in REQUIRES_NEW so it commits before the Stripe call
        // and is visible to concurrent service calls with the same command.
        savePending(command, idempotencyKey); // @Transactional(REQUIRES_NEW, DataSource)

        // Stripe — on @Retryable attempt 2, content-hash key = same as attempt 1,
        // Stripe returns ch_A from its idempotency cache. No ch_B.
        StripeCharge charge = stripeClient.createCharge(
            command.getAmountCents(), command.getCustomerId(), idempotencyKey);

        kafkaTemplate.send("payment-results", command.getOrderId(),
            new PaymentResult(command.getOrderId(), charge.getId(), idempotencyKey));

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

An alternative approach for teams that want to preserve UUID.randomUUID() for other reasons: compute the UUID in the @KafkaListener method (outside the @Retryable + @Transactional boundary) and pass it into processBilling() as a parameter. This scopes the UUID generation to the Kafka record delivery, not to the @Retryable attempt boundary. However, this approach does not survive a Kafka-level re-delivery of the original record (e.g., if the @KafkaListener itself fails and DefaultErrorHandler seeks back), so the content-hash approach is more robust.

Mode 3: ChainedKafkaTransactionManager sequential commit — JDBC commits before Kafka fails — consumer offset not advanced — re-delivery with UUID_B

Spring Kafka’s ChainedKafkaTransactionManager (CKTM) was designed to coordinate a KafkaTransactionManager with one or more additional PlatformTransactionManager instances (typically a DataSourceTransactionManager) so that @KafkaListener methods can participate in both a Kafka transaction (covering the outbound kafkaTemplate.send() and the consumer offset advance via sendOffsetsToTransaction()) and a JDBC transaction (covering DB writes) within a single @Transactional boundary. CKTM was deprecated in Spring Kafka 2.7, with the recommendation to use @Transactional on the listener method with the KafkaTransactionManager directly and a separate @Transactional(DataSource) for DB operations. It is still widely deployed in codebases that have not been updated.

The commit mechanism in CKTM is sequential, not atomic. CKTM commits its chained transaction managers in the following order: the additional managers registered in the constructor (the DataSourceTransactionManager, index 1) are committed first, then the primary KafkaTransactionManager (index 0) is committed. This ordering is deliberate: it ensures that if the Kafka commit fails, the JDBC data is durable, and a subsequent re-delivery can detect the prior processing via the JDBC data.

The failure mode occurs when the developer uses UUID.randomUUID() as the Stripe idempotency key — a random key that is NOT the same across re-deliveries. In that case, the CKTM commit ordering produces a double charge:

// CKTM configuration: DataSourceTransactionManager committed first, Kafka second
@Bean
ChainedKafkaTransactionManager<String, Object> chainedKafkaTxManager(
    KafkaTransactionManager<String, Object> kafkaTxManager,
    DataSourceTransactionManager dataSourceTxManager) {
    // Primary = kafkaTxManager. Additional = dataSourceTxManager.
    // Commit order: dataSourceTxManager first, then kafkaTxManager.
    return new ChainedKafkaTransactionManager<>(kafkaTxManager, dataSourceTxManager);
}

@KafkaListener(
    topics = "payment-commands",
    groupId = "billing-group",
    containerFactory = "chainedTxContainerFactory"
)
@Transactional("chainedKafkaTransactionManager")
void handlePayment(ConsumerRecord<String, PaymentCommand> record) {
    PaymentCommand command = record.value();
    String idempotencyKey = UUID.randomUUID().toString(); // UUID_A on attempt 1

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

    // Kafka outbound message (inside KafkaTransactionManager transaction)
    kafkaTemplate.send("payment-results", command.getOrderId(),
        new PaymentResult(command.getOrderId(), charge.getId(), idempotencyKey));

    // DB write (inside DataSourceTransactionManager transaction)
    paymentRecordRepository.save(
        new PaymentRecord(command.getOrderId(), idempotencyKey, charge.getId(), "COMPLETED")
    );
}

Step-by-step failure trace

Attempt 1 — partial commit scenario:

1. CKTM opens two transactions: a DataSourceTransactionManager transaction (JDBC TX_1) and a KafkaTransactionManager transaction (Kafka TX_1, calling KafkaProducer.beginTransaction()).
2. UUID.randomUUID() → UUID_A.
3. stripeClient.createCharge(5000, “cust_1”, UUID_A) → ch_A committed on Stripe (HTTP 200).
4. kafkaTemplate.send(“payment-results”, …) → message buffered inside Kafka TX_1.
5. paymentRecordRepository.save() → INSERT inside JDBC TX_1. The row is uncommitted at this point.
6. The listener method returns normally. CKTM’s commit() is invoked.
7. CKTM commits the DataSourceTransactionManager first: JDBC TX_1 commits. The payment_records row with idempotencyKey = UUID_A and chargeId = ch_A is permanently committed to the database.
8. CKTM commits the KafkaTransactionManager next. This involves: (a) KafkaProducer.sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata>, groupMetadata) — atomically binding the consumer offset advance (input partition: offset 42+1=43) with the output message on payment-results to Kafka TX_1; (b) KafkaProducer.commitTransaction(). During commitTransaction(), a TimeoutException is thrown — the Kafka broker’s __consumer_offsets partition leader is unavailable during an election that coincides with this commit attempt.
9. The Kafka transaction is aborted by the coordinator. The buffered message on payment-results is NOT committed (no read_committed consumer sees it). The consumer offset for payment-commands partition 0 remains at the last successfully committed offset (before offset 42). The JDBC transaction is not rolled back — it was committed in step 7 and is permanently durable.
10. The TimeoutException from the Kafka commit propagates out of CKTM’s commit(). Spring Kafka’s container handles it as a listener failure. The container calls consumer.seek(new TopicPartition(“payment-commands”, 0), 42) to reset the partition position.

Attempt 2 — re-delivery:

1. On re-poll, the container delivers the same ConsumerRecord at partition 0, offset 42, to handlePayment().
2. CKTM opens new JDBC TX_2 and Kafka TX_2.
3. UUID.randomUUID() → UUID_B.
4. stripeClient.createCharge(5000, “cust_1”, UUID_B) → Stripe looks up UUID_B: it has never been seen. Stripe creates ch_B. ch_B is committed alongside ch_A.

At this point: the database has a payment_records row for UUID_A / ch_A (from attempt 1, JDBC TX_1). Stripe has ch_A and ch_B. The customer has been charged twice.

Developer misconception

“ChainedKafkaTransactionManager provides an atomic commit for both Kafka and JDBC. If the Kafka commit fails, the entire CKTM transaction is rolled back — the JDBC write never happened and a clean retry is safe. CKTM is specifically designed for this scenario.”

CKTM does not provide two-phase commit. It provides sequential, best-effort commit. The CKTM Javadoc states explicitly: “Since there is no distributed transaction coordinator, if there is an error on the commit, the first transaction manager may have committed but the second will roll back (effectively a best-effort 1PC).” When the JDBC transaction commits (step 7 above), the commit is permanent and irrevocable. There is no mechanism in JDBC to roll back a committed transaction. CKTM’s rollback() method can call dataSourceTransactionManager.rollback(), but that only affects an uncommitted JDBC transaction — it cannot undo a JDBC transaction that has already committed. In the failure scenario above, the JDBC transaction had already committed before the Kafka commit threw.

The reason this failure mode is more subtle than the others in this post: the CKTM partial commit is not triggered by an application-logic exception (like DataAccessException) but by an infrastructure failure (Kafka broker leader election, network partition to broker, transaction.timeout.ms expiry). These failures are rare in normal operation, making the bug invisible in load testing and hard to reproduce in integration tests without deliberate broker fault injection.

Fix

The content-hash key is the correct and complete fix for the CKTM partial commit scenario. When attempt 2 re-delivers the same PaymentCommand, the content-hash produces the same UUID_A. paymentRecordRepository.findByIdempotencyKey(UUID_A) finds the JDBC-committed row from attempt 1 (with chargeId = ch_A). The guard returns early, sends the existing ch_A charge ID to the Kafka output topic (inside the new Kafka TX_2), and commits — this time without calling Stripe. No ch_B is created.

@KafkaListener(
    topics = "payment-commands",
    groupId = "billing-group",
    containerFactory = "chainedTxContainerFactory"
)
@Transactional("chainedKafkaTransactionManager")
void handlePayment(ConsumerRecord<String, PaymentCommand> record) {
    PaymentCommand command = record.value();

    // Content-hash key: same value on any CKTM partial-commit re-delivery
    String idempotencyKey = contentHashKey(command);

    // Guard: find existing record from a prior JDBC commit (CKTM partial commit scenario)
    Optional<PaymentRecord> existing =
        paymentRecordRepository.findByIdempotencyKey(idempotencyKey);

    if (existing.isPresent() && existing.get().getChargeId() != null) {
        // CKTM partial commit: JDBC row with ch_A exists but Kafka TX failed.
        // Re-send the result message to complete the consume-transform-produce pipeline.
        // This Kafka send is inside the new Kafka TX_2, which will commit successfully.
        kafkaTemplate.send("payment-results", command.getOrderId(),
            new PaymentResult(command.getOrderId(), existing.get().getChargeId(), idempotencyKey));
        return;
    }

    if (existing.isPresent()) {
        // PENDING row from a concurrent consumer — wait or skip
        return;
    }

    // PENDING row (committed inside REQUIRES_NEW before the Stripe call)
    savePending(command, idempotencyKey);

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

    kafkaTemplate.send("payment-results", command.getOrderId(),
        new PaymentResult(command.getOrderId(), charge.getId(), idempotencyKey));

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

The PENDING row’s REQUIRES_NEW ensures it commits before the Stripe call even in the CKTM context (where the outer transaction is the CKTM). On the re-delivery, the guard finds the COMPLETED row (because JDBC TX_1 committed in step 7 of the failure trace), re-sends the output message to Kafka, and commits Kafka TX_2 — this time without the broker infrastructure failure that caused the original commit to fail.

Teams running CKTM should note that this JDBC-first commit ordering is intentional: if the Kafka commit fails, the JDBC data (specifically the payment_records row with the charge ID) is available to the re-delivery to reconstruct the correct Stripe charge reference. The content-hash key makes the JDBC row findable on re-delivery. Without the content-hash key, the JDBC row (keyed on UUID_A) is invisible to the re-delivery that generates UUID_B.

Comparison table

Integration tests

Each mode can be reproduced deterministically in a @SpringBootTest with Testcontainers (Kafka 3.x, PostgreSQL 15) and WireMock. The key test assertion in all three modes is the same: verify that Stripe received the same Idempotency-Key header on every attempt for the same PaymentCommand.

Mode 1 (DefaultErrorHandler re-delivery) test setup

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"payment-commands", "payment-results"})
class DefaultErrorHandlerStripeTest {

    @Autowired KafkaTemplate<String, PaymentCommand> commandTemplate;
    @Autowired PaymentRecordRepository paymentRecordRepository;

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

    @BeforeEach
    void setup() {
        stripe.start();
        stripe.stubFor(post(urlPathEqualTo("/v1/charges"))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_test_001\",\"status\":\"succeeded\"}")));
    }

    @AfterEach void tearDown() { stripe.stop(); }

    @Test
    void defaultErrorHandlerRetry_randomUuid_generatesUuidBOnRedelivery() throws Exception {
        // Force first save() to throw, second to succeed
        // (use a spy or a toggle in the repository mock)
        when(paymentRecordRepository.save(any()))
            .thenThrow(new DataAccessException("simulated constraint violation") {})
            .thenCallRealMethod();

        commandTemplate.send("payment-commands", "order_1",
            new PaymentCommand("cust_1", "order_1", 5000));

        // Wait for DefaultErrorHandler to re-deliver (FixedBackOff 1000ms + processing)
        Thread.sleep(3000);

        // Assert: Stripe received 2 requests with DIFFERENT idempotency keys
        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(2);

        String key1 = stripeRequests.get(0).getHeader("Idempotency-Key");
        String key2 = stripeRequests.get(1).getHeader("Idempotency-Key");
        assertThat(key1).isNotEqualTo(key2); // BUG: two distinct keys, two charges
    }

    @Test
    void defaultErrorHandlerRetry_contentHashKey_sameKeyOnRedelivery() throws Exception {
        when(paymentRecordRepository.save(any()))
            .thenThrow(new DataAccessException("simulated constraint violation") {})
            .thenCallRealMethod();

        commandTemplate.send("payment-commands", "order_1",
            new PaymentCommand("cust_1", "order_1", 5000));
        Thread.sleep(3000);

        // Assert: Stripe received 2 requests with THE SAME idempotency key
        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/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); // FIX: same content-hash key on both deliveries

        // Assert: only 1 payment_records row in the database
        assertThat(paymentRecordRepository.findByOrderId("order_1")).hasSize(1);
    }
}

Mode 2 (@Retryable + @Transactional(kafkaTxManager)) test setup

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"payment-commands", "payment-results"})
class RetryableKafkaTxManagerStripeTest {

    @Autowired BillingService billingService; // @Retryable + @Transactional(kafkaTxManager)
    @Autowired PaymentRecordRepository paymentRecordRepository;

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

    @BeforeEach void setup() {
        stripe.start();
        stripe.stubFor(post(urlPathEqualTo("/v1/charges"))
            .willReturn(aResponse().withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_test_002\",\"status\":\"succeeded\"}")));
    }

    @AfterEach void tearDown() { stripe.stop(); }

    @Test
    void retryable_randomUuid_generatesNewUuidOnEachAttempt() {
        when(paymentRecordRepository.save(any()))
            .thenThrow(new DataAccessException("deadlock") {})  // attempt 1: fail
            .thenThrow(new DataAccessException("deadlock") {})  // attempt 2: fail
            .thenCallRealMethod();                               // attempt 3: succeed

        billingService.processBilling(
            new PaymentCommand("cust_1", "order_1", 5000));

        // @Retryable attempted 3 times — each time UUID.randomUUID() was called
        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(3);

        // Assert: all 3 Stripe requests had DIFFERENT idempotency keys
        Set<String> keys = stripeRequests.stream()
            .map(r -> r.getHeader("Idempotency-Key"))
            .collect(Collectors.toSet());
        assertThat(keys).hasSize(3); // BUG: 3 distinct keys = 3 charges on Stripe
    }

    @Test
    void retryable_contentHashKey_sameKeyOnAllAttempts() {
        when(paymentRecordRepository.save(any()))
            .thenThrow(new DataAccessException("deadlock") {})
            .thenCallRealMethod();

        billingService.processBilling(
            new PaymentCommand("cust_1", "order_1", 5000));

        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/v1/charges")));

        // Stripe called twice (attempts 1 and 2), but same content-hash key on both
        Set<String> keys = stripeRequests.stream()
            .map(r -> r.getHeader("Idempotency-Key"))
            .collect(Collectors.toSet());
        assertThat(keys).hasSize(1); // FIX: single content-hash key across @Retryable attempts

        assertThat(paymentRecordRepository.findByOrderId("order_1")).hasSize(1);
    }
}

Mode 3 (CKTM partial commit) test setup

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"payment-commands", "payment-results"})
class ChainedKafkaTxManagerStripeTest {

    @Autowired KafkaTemplate<String, PaymentCommand> commandTemplate;
    @Autowired PaymentRecordRepository paymentRecordRepository;

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

    @BeforeEach void setup() { stripe.start(); /* stubFor /v1/charges ... */ }
    @AfterEach void tearDown() { stripe.stop(); }

    @Test
    void cktmPartialCommit_randomUuid_jdbcCommittedKafkaFails_redeliveryCreatesChB()
            throws Exception {
        // Simulate CKTM partial commit: intercept KafkaTransactionManager.commit()
        // to throw TimeoutException on first invocation (after JDBC has committed).
        // Second invocation succeeds.
        KafkaTransactionManager<?, ?> kafkaTxManager = ...; // get from context
        KafkaTransactionManager<?, ?> spyTxManager = spy(kafkaTxManager);
        doThrow(new KafkaException("simulated broker timeout"))
            .doCallRealMethod()
            .when(spyTxManager).doCommit(any());
        // Wire spy into the CKTM for this test...

        commandTemplate.send("payment-commands", "order_1",
            new PaymentCommand("cust_1", "order_1", 5000));
        Thread.sleep(4000); // allow for CKTM re-delivery

        // Assert: Stripe received 2 requests with DIFFERENT keys (UUID_A and UUID_B)
        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(2);
        assertThat(stripeRequests.get(0).getHeader("Idempotency-Key"))
            .isNotEqualTo(stripeRequests.get(1).getHeader("Idempotency-Key")); // BUG
    }

    @Test
    void cktmPartialCommit_contentHashKey_redeliveryFindsJdbcRow_skipsStripe()
            throws Exception {
        // Same CKTM spy setup: Kafka commit fails on first attempt, succeeds on second

        commandTemplate.send("payment-commands", "order_1",
            new PaymentCommand("cust_1", "order_1", 5000));
        Thread.sleep(4000);

        // Assert: Stripe called exactly ONCE (attempt 1 only)
        // — attempt 2 found the JDBC-committed row, skipped Stripe
        stripe.verify(1, postRequestedFor(urlPathEqualTo("/v1/charges")));

        // Assert: both Stripe calls (if any) used the same content-hash key
        List<LoggedRequest> stripeRequests =
            stripe.findAll(postRequestedFor(urlPathEqualTo("/v1/charges")));
        if (stripeRequests.size() == 2) {
            assertThat(stripeRequests.get(0).getHeader("Idempotency-Key"))
                .isEqualTo(stripeRequests.get(1).getHeader("Idempotency-Key")); // FIX
        }

        // Assert: payment-results topic received exactly one message with ch_A
        // (from attempt 2's guard path: re-sent the existing charge ID to Kafka)
        // ... consume from payment-results and assert single PaymentResult with ch_test_001
    }
}

Summary

All three failure modes in this post share the same root cause as the Micronaut Kafka EOS post: the developer believes that some Spring or Kafka infrastructure mechanism provides atomicity that covers both Kafka operations and external HTTP calls to Stripe. It does not. DefaultErrorHandler’s seek() + re-poll is a Kafka offset operation, not a partial-retry mechanism that skips already-executed lines. @Retryable’s MethodInvocation.proceed() is a full AOP chain re-invocation, not a selective re-execution of the failing line. ChainedKafkaTransactionManager’s commit is a sequential two-step, not an atomic 2PC.

The fix is the same in all three modes: derive the Stripe idempotency key from the stable fields of the PaymentCommand (a SHA-256 content hash of customerId + orderId + amountCents), not from UUID.randomUUID(). The content hash produces the same key on every re-delivery or retry of the same logical billing event, regardless of whether the re-execution was triggered by DefaultErrorHandler, @Retryable, or a CKTM partial commit re-delivery. Stripe’s 24h idempotency cache returns the original charge on any re-delivery within the window. The PENDING row pattern serializes concurrent consumer instances that may process the same record in a rebalance window. Together, these two patterns make the listener method safe under all three failure modes described in this post and under the three Micronaut EOS failure modes described in the prior post.

One note on the CKTM partial commit mode specifically: the fix turns CKTM’s JDBC-first commit ordering from a liability into an asset. The JDBC-committed payment_records row, keyed on the content hash, becomes the recovery checkpoint for the re-delivery. The re-delivery finds the row, extracts the existing chargeId = ch_A, sends the result to the Kafka output topic (now inside a new Kafka transaction that is not subject to the original broker failure), and returns without calling Stripe. The content-hash key is what makes this recovery deterministic — without it, the JDBC row from attempt 1 (keyed on UUID_A) is invisible to the re-delivery that generates UUID_B.