Micronaut Kafka and Stripe Integration: How @KafkaListener with @Retryable Re-enters the Method Body After billingRepository.save() Fails Generating UUID_B While ch_A Stays Committed on Stripe, @KafkaClient Inside @Transactional Delivers the Message to the Broker Before Rollback Causing a Duplicate Consumer Invocation That Double-Charges with UUID_B, and batch=true Redelivers the Entire Batch After Mid-Batch Failure Re-processing Already-Charged Records with New Random Keys

The prior Micronaut Data JDBC post in this series covered three failure modes at the SQL data-access layer: saveAll() batch retry regenerating a UUID per item when the batch INSERT fails after Stripe charges committed, a custom @Query ON CONFLICT DO NOTHING guard being bypassed by UUID.randomUUID() idempotency keys, and READ COMMITTED isolation allowing two concurrent findByIdempotencyKey() checks to both return empty before either request commits. All three failures are rooted in the data layer. This post moves to the messaging layer and covers three failure modes specific to Micronaut’s @KafkaListener and @KafkaClient abstractions. Each mode produces the same observable: Stripe records two Charge or PaymentIntent objects for the same logical billing event with distinct Idempotency-Key header values, and the customer’s payment method is debited twice.

The three failure modes in this post are mechanically distinct from each other and from all prior Micronaut posts in this series. The first is an AOP-layer failure: Micronaut’s RetryInterceptor re-invokes the @KafkaListener method body via MethodInvocationContext.proceed() on each retry attempt, regenerating the idempotency key. The second is a transport-layer failure: @KafkaClient’s Kafka producer sends messages to the broker via the Kafka NetworkClient, which is not a JTA resource and does not participate in Micronaut’s DataSourceTransactionManager — when @Transactional rolls back, only JDBC writes are undone, and a message sent inside the transaction stays on the broker. The third is a Kafka offset-commit failure: @KafkaListener(batch=true) does not commit the consumer group offset until the entire batch method returns successfully, so a mid-batch exception causes Kafka to redeliver the full batch from the last committed offset, re-processing records that were already charged on Stripe in the first pass.

For the tests in this post, the stack is: Micronaut 4.x, micronaut-kafka, Micronaut Data JPA (for repository layer), PostgreSQL 15 via Testcontainers, Apache Kafka via Testcontainers, WireMock for Stripe API stubs, and JUnit 5 with @MicronautTest.

Background: Micronaut Kafka @KafkaClient and @KafkaListener mechanics

Micronaut Kafka is a compile-time integration layer over Apache Kafka’s Java client (kafka-clients). At build time, Micronaut’s annotation processor generates implementations for interfaces annotated with @KafkaClient and registers @KafkaListener-annotated beans as consumers in a managed consumer poll loop. The framework wraps the Kafka KafkaProducer and KafkaConsumer types internally and exposes them through Micronaut’s DI container as singletons.

The @KafkaClient annotation on an interface tells Micronaut to generate a proxy implementation. Each method on the interface corresponds to a Kafka producer send operation. The generated implementation calls KafkaProducer.send(ProducerRecord, Callback) on the underlying Kafka producer. By default, the send is synchronous from the method caller’s perspective: the generated code calls Future<RecordMetadata>.get() and blocks until the broker acknowledges the message with acks=1 (the Kafka default). Methods that return Single<RecordMetadata> or Publisher<RecordMetadata> return without blocking and complete the reactive type when the broker acknowledges.

The @KafkaListener annotation on a class (or a method within a singleton) registers a Kafka consumer that participates in the named consumer group. Micronaut Kafka starts a dedicated consumer thread per listener. The thread runs the Kafka consumer poll loop: KafkaConsumer.poll(Duration) returns a batch of ConsumerRecord objects, and Micronaut invokes the annotated listener method for each record (or for the entire batch when batch=true). After the listener method returns, Micronaut commits the consumer group offset depending on the configured ackMode. The default ackMode is AUTO, which commits after each successful method invocation (for single-record listeners) or after the entire batch method returns (for batch=true listeners).

The critical architectural distinction for Stripe integration: neither @KafkaClient sends nor @KafkaListener offset commits participate in Micronaut’s database transaction management. Micronaut’s @Transactional is backed by DataSourceTransactionManager, which manages JDBC Connection objects and their auto-commit flag. Kafka’s KafkaProducer and KafkaConsumer use their own NetworkClient to communicate with the Kafka broker over TCP. There is no XA coordinator, no two-phase commit protocol, and no mechanism for Micronaut’s transaction manager to include Kafka operations in a JDBC transaction rollback.

Mode 1: Micronaut @Retryable on @KafkaListener method — RetryInterceptor re-invokes the method body via proceed() — UUID.randomUUID() at method entry generates UUID_B — ch_B committed alongside committed ch_A

The developer writes a payment consumer that receives payment commands from a Kafka topic, calls Stripe to create a charge, and saves the result to the database. Knowing that database writes can fail transiently (network hiccups, connection pool exhaustion), the developer adds Micronaut’s @Retryable annotation to the listener method. The Stripe idempotency key is generated at the top of the method with UUID.randomUUID(). The developer’s reasoning: if the database write fails and @Retryable triggers, the @Transactional rollback ensures the database is clean, so retrying is safe.

// PaymentCommand.java — Kafka message payload
public record PaymentCommand(
    String customerId,
    String orderId,
    int amountCents
) {}

// PaymentConsumer.java — unsafe: UUID.randomUUID() at method entry + @Retryable on @KafkaListener
@Singleton
public class PaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    public PaymentConsumer(PaymentRecordRepository paymentRecordRepository,
                           StripeClient stripeClient) {
        this.paymentRecordRepository = paymentRecordRepository;
        this.stripeClient = stripeClient;
    }

    @Retryable(attempts = "3", delay = "1s")
    @KafkaListener(groupId = "payment-processor")
    @Topic("payment-commands")
    @Transactional
    public void onPaymentCommand(PaymentCommand command) {

        // UUID generated at method entry — new value on EVERY @Retryable invocation
        String idempotencyKey = UUID.randomUUID().toString();

        // Step 1: Call Stripe — ch_A committed on Stripe on attempt 1 with key_A
        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey         // key_A on attempt 1, key_B on @Retryable attempt 2
        );

        // Step 2: Save result to DB — may throw DataAccessException
        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);        // key_A
        record.setCustomerId(command.customerId());
        record.setOrderId(command.orderId());
        record.setAmountCents(command.amountCents());
        record.setStripeChargeId(charge.chargeId());     // ch_A
        record.setStatus("COMPLETED");
        paymentRecordRepository.save(record);
        // DataAccessException thrown here (e.g. transient connection loss, UNIQUE violation)
        // @Transactional rolls back: payment_records INSERT — REVERTED
        // Stripe ch_A: NOT reverted — Stripe committed it when it returned HTTP 200
    }
}

The developer’s mental model is that @Retryable is an error-handling mechanism that operates after the @Transactional boundary has cleaned up. The database rollback is complete, the record of the first attempt is gone, so the retry is operating on a clean slate. The developer concludes that UUID.randomUUID() in the retry is safe because there is nothing to deduplicate against.

This reasoning is incomplete at the AOP interceptor level and incorrect at the external-API level. At the AOP level, Micronaut’s interceptors are applied in the order they are declared on the method. When a method is annotated with both @Retryable and @Transactional, the compiled proxy wraps them such that RetryInterceptor is the outermost interceptor and TransactionalInterceptor is inner. The execution order is: RetryInterceptor.proceed() → TransactionalInterceptor.proceed() → actual method body. When paymentRecordRepository.save() throws, the exception propagates through TransactionalInterceptor, which issues the rollback, and then propagates to RetryInterceptor, which catches it and schedules a retry. The retry re-calls proceed() on the chain, which re-invokes TransactionalInterceptor (opening a new JDBC transaction) and then re-invokes the actual method body from line 1. Line 1 is String idempotencyKey = UUID.randomUUID().toString(). The new UUID value is assigned on each method re-entry.

At the external-API level, the sequence of events on attempt 1 is:

1. Kafka consumer delivers the PaymentCommand for orderId=ORDER-1 to the listener method.
2. @Retryable interceptor wraps execution. @Transactional interceptor opens a new JDBC transaction.
3. UUID.randomUUID() at line 1 generates key_A.
4. stripeClient.createCharge(customerId, amountCents, key_A) is called. Stripe receives the POST, processes the charge, and returns HTTP 200 with {"id":"ch_A",...}. ch_A is committed on Stripe. Stripe stores key_A in its idempotency log for 24 hours.
5. paymentRecordRepository.save(record) throws DataAccessException (for example, the JDBC connection to PostgreSQL was interrupted mid-write by a failover).
6. @Transactional interceptor catches the exception, calls connection.rollback(), and releases the connection. The payment_records INSERT is undone. The database has no row with key_A or ch_A.
7. @Retryable interceptor receives the exception. It waits 1 second and re-calls proceed().
8. Attempt 2 begins. A new JDBC transaction is opened. Line 1: UUID.randomUUID() generates key_B.
9. stripeClient.createCharge(customerId, amountCents, key_B) is called. Stripe checks its idempotency log: key_B has never been seen (key_A is in the log, but key_B is a distinct 128-bit random string). Stripe treats this as a new charge request. ch_B is committed on Stripe. The customer’s card is debited a second time.
10. If paymentRecordRepository.save() succeeds on attempt 2, the database records key_B and ch_B. The application has a record of ch_B. ch_A is committed on Stripe with no application-side record.

Why @Retryable on @KafkaListener is a common pattern despite this risk

Kafka consumers are long-lived polling threads. Without retry logic, a transient database failure would either skip the message (if the consumer has auto-commit configured) or block the entire consumer partition indefinitely (if the exception propagates). Adding @Retryable to the listener method is a natural defensive response: handle transient failures without losing messages and without stalling the consumer. The pattern is correct in intent. The mistake is placing the idempotency key generation inside the retryable method body rather than deriving it from the message content before any retry boundary.

Fix: content-hash key from stable Kafka message fields

The Kafka message carries the billing event’s identifying fields: customerId, orderId, and amountCents. These values are stable across every redelivery of the same message (the Kafka broker stores them). A content-hash key derived from these fields is deterministic: every invocation of the consumer for the same PaymentCommand message produces the same idempotency key. When @Retryable re-invokes the method, the content-hash key matches the key used in attempt 1. Stripe looks up the key in its idempotency log, finds ch_A, and returns ch_A without creating a new charge.

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

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    // No @Retryable here — retry is handled at the Kafka offset level (message redelivery)
    @KafkaListener(groupId = "payment-processor")
    @Topic("payment-commands")
    @Transactional
    public void onPaymentCommand(PaymentCommand command) {

        // Deterministic key — same inputs always produce the same key
        String idempotencyKey = buildContentHash(
            command.customerId(),
            command.orderId(),
            String.valueOf(command.amountCents())
        );

        // Check for prior completed processing (handles: message redelivery after consumer restart)
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            // Already processed — idempotent return
            // Kafka will commit the offset after this method returns successfully
            return;
        }

        // Insert PENDING row BEFORE calling Stripe
        // If a concurrent consumer instance also received this message (split-brain rare case),
        // the UNIQUE constraint on idempotencyKey serializes access
        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);
        record.setCustomerId(command.customerId());
        record.setOrderId(command.orderId());
        record.setAmountCents(command.amountCents());
        record.setStatus("PENDING");
        try {
            paymentRecordRepository.saveAndFlush(record);
        } catch (DataIntegrityViolationException e) {
            // Concurrent consumer beat us to the PENDING row — wait for their result
            // (The Kafka offset will not be committed, message will be redelivered after
            // the concurrent consumer commits the offset from its own successful processing)
            throw e;
        }

        // Stripe charge — content-hash key ensures Stripe deduplicates across redeliveries
        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey         // same key on every invocation for this orderId
        );

        // Update PENDING to COMPLETED
        record.setStripeChargeId(charge.chargeId());
        record.setStatus("COMPLETED");
        paymentRecordRepository.update(record);
        // @Transactional commits here — then Kafka auto-commits the offset
    }

    private String buildContentHash(String... inputs) {
        String raw = String.join("|", inputs);
        try {
            MessageDigest sha256 = MessageDigest.getInstance("SHA-256");
            byte[] hash = sha256.digest(raw.getBytes(StandardCharsets.UTF_8));
            return "kb_" + HexFormat.of().formatHex(hash).substring(0, 40);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 not available", e);
        }
    }
}

Note that @Retryable has been removed from the listener method in the fixed version. Retry for transient failures is handled at the Kafka level: if the consumer throws, the offset is not committed, and the broker will redeliver the message on the next poll. The content-hash key ensures that redelivery is safe regardless of whether the previous attempt called Stripe or not. If the previous attempt called Stripe and committed ch_A, the redelivery will find ch_A via Stripe’s idempotency cache (same key) and the existsByIdempotencyKey check will find the COMPLETED row (if the DB write also committed) or the PENDING row (if the DB write was mid-flight when the consumer failed).

Mode 2: @KafkaClient send inside @Transactional — Kafka producer is not a JTA resource — message reaches broker before rollback — @Retryable sends a second message — consumer processes both with UUID.randomUUID() — ch_A and ch_B both committed on Stripe

A common pattern in event-driven architectures is to publish a domain event to Kafka from within a service method that also writes to the database. The developer wraps both operations in @Transactional and assumes that if the database write fails and the transaction rolls back, the Kafka message is also rolled back. This assumption is incorrect for Micronaut’s default @KafkaClient. The developer compounds the risk by placing the Stripe charge in the downstream consumer, where UUID.randomUUID() generates a new key on each invocation.

// PaymentCommandClient.java — @KafkaClient interface (Micronaut generates the implementation)
@KafkaClient
public interface PaymentCommandClient {
    @Topic("payment-commands")
    void sendPaymentCommand(PaymentCommand command);
}

// OrderService.java — unsafe: @KafkaClient send inside @Transactional
@Singleton
public class OrderService {

    private final OrderRepository orderRepository;
    private final PaymentCommandClient paymentCommandClient;

    public OrderService(OrderRepository orderRepository,
                        PaymentCommandClient paymentCommandClient) {
        this.orderRepository = orderRepository;
        this.paymentCommandClient = paymentCommandClient;
    }

    @Retryable(attempts = "3", delay = "2s", includes = DataAccessException.class)
    @Transactional
    public void processOrder(String orderId, String customerId, int amountCents) {

        // Step 1: Save order to DB
        Order order = new Order(orderId, customerId, amountCents, "PROCESSING");
        orderRepository.save(order);

        // Step 2: Send payment command to Kafka
        // MISCONCEPTION: developer believes this send is part of the @Transactional boundary
        // and will be rolled back if the transaction rolls back
        paymentCommandClient.sendPaymentCommand(
            new PaymentCommand(customerId, orderId, amountCents)
        );
        // At this point, the Kafka broker has acknowledged cmd1 (acks=1 default).
        // cmd1 is on the broker's log at offset N. This is a committed fact
        // that has nothing to do with the JDBC transaction's state.

        // Step 3: Update order status — may fail with DataAccessException
        order.setStatus("PAYMENT_PENDING");
        orderRepository.update(order); // may throw DataAccessException
        // @Transactional rolls back:
        //   - order INSERT (Step 1)       — REVERTED
        //   - order UPDATE (Step 3)       — REVERTED
        //   - Kafka cmd1 (Step 2)         — NOT REVERTED (broker has it)
    }
}

// PaymentConsumer.java — unsafe downstream consumer with UUID.randomUUID()
@Singleton
public class PaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

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

        // New UUID on each consumer invocation — does not deduplicate across messages
        String idempotencyKey = UUID.randomUUID().toString();

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

        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);
        record.setCustomerId(command.customerId());
        record.setOrderId(command.orderId());
        record.setStripeChargeId(charge.chargeId());
        record.setStatus("COMPLETED");
        paymentRecordRepository.save(record);
    }
}

The developer’s reasoning: “@Transactional wraps the entire processOrder method body. Both the orderRepository.save() call and the paymentCommandClient.sendPaymentCommand() call happen inside the transaction scope. If the transaction rolls back, all operations inside the scope are reversed. The Kafka send is inside the scope, so it is reversed.”

This reasoning is wrong in a way that is specific to how @KafkaClient is implemented. Micronaut’s @Transactional is backed by DataSourceTransactionManager. This transaction manager manages a JDBC Connection: it sets connection.setAutoCommit(false) at the start of the transaction and calls connection.rollback() or connection.commit() at the end. The scope of DataSourceTransactionManager’s rollback is exactly the JDBC operations performed through the managed Connection: the SQL INSERT from orderRepository.save() and the SQL UPDATE from orderRepository.update().

@KafkaClient’s generated implementation calls KafkaProducer.send(ProducerRecord, Callback). The KafkaProducer internally uses a NetworkClient that communicates with the Kafka broker over a TCP connection managed by Kafka’s own I/O thread. This NetworkClient is not registered with Micronaut’s transaction manager. There is no XAResource registration, no two-phase commit participant, and no callback mechanism between DataSourceTransactionManager and KafkaProducer. When paymentCommandClient.sendPaymentCommand(cmd1) is called and the Kafka broker returns an acknowledgment, cmd1 is committed to the broker’s log at offset N. This commitment is permanent and independent of whatever the JDBC transaction does afterward.

The sequence of events that produces the double charge:

1. Attempt 1: processOrder("ORDER-1", "cus_1", 9900) is called. @Retryable wraps execution. @Transactional opens JDBC transaction T1.
2. T1: orderRepository.save(order) — SQL INSERT INTO orders executed over the managed JDBC connection. Row exists in the DB but is not yet visible outside T1 (uncommitted).
3. T1 (inside): paymentCommandClient.sendPaymentCommand(cmd1) — Micronaut calls KafkaProducer.send(paymentCommandsTopic, cmd1) on the Kafka producer. The producer’s NetworkClient sends the message to the broker leader. The broker appends cmd1 to the payment-commands topic log at offset 42 and sends an acknowledgment. send() returns with RecordMetadata(offset=42). cmd1 is committed on the broker. This is irreversible.
4. T1: orderRepository.update(order) — SQL UPDATE orders SET status=‘PAYMENT_PENDING’ — throws DataAccessException (for example, the orders table has a row-level lock from a concurrent admin update on this order ID).
5. T1 rolls back: the INSERT from step 2 is undone. The UPDATE from step 4 is undone. No JDBC changes are visible. cmd1 on the broker is NOT rolled back.
6. @Retryable waits 2 seconds and re-invokes processOrder. @Transactional opens JDBC transaction T2.
7. T2: orderRepository.save(order) — INSERT INTO orders for ORDER-1 (succeeds, since T1’s INSERT was rolled back).
8. T2 (inside): paymentCommandClient.sendPaymentCommand(cmd2) — Kafka producer sends a second PaymentCommand for ORDER-1 to the payment-commands topic. Broker appends at offset 43. cmd2 is committed on the broker.
9. T2: orderRepository.update(order) succeeds. T2 commits.
10. The payment-commands topic now has cmd1 at offset 42 and cmd2 at offset 43, both for ORDER-1, both with identical customerId and amountCents.
11. The PaymentConsumer consumer group polls the topic. It receives cmd1 (offset 42) and invokes onPaymentCommand(cmd1). UUID.randomUUID() generates key_1. stripeClient.createCharge(cus_1, 9900, key_1) is called. ch_A committed on Stripe.
12. Consumer polls again. It receives cmd2 (offset 43) and invokes onPaymentCommand(cmd2). UUID.randomUUID() generates key_2 (different from key_1). stripeClient.createCharge(cus_1, 9900, key_2) is called. Stripe checks its idempotency log: key_2 has never been seen. ch_B committed on Stripe. The customer’s card is debited twice.

Why this pattern is attractive despite the risk

The transactional outbox pattern — where an outbox record written to the DB in the same transaction serves as the Kafka message source — is widely recommended but requires a separate relay process, increased DB write volume, and careful handling of at-least-once delivery from the outbox. Wrapping both the DB write and the Kafka send in @Transactional feels like a simpler alternative. Developers who have worked primarily with fully transactional message brokers (like ActiveMQ with XA), or with Spring’s ChainedTransactionManager (which does not provide true atomicity but at least couples the two operations), may carry over an expectation of transactional Kafka sends that does not apply to Micronaut’s default @KafkaClient.

Fix: transactional outbox pattern + content-hash key in the consumer

The safe pattern has two parts. First, replace the direct @KafkaClient send with a database outbox record written inside the same @Transactional as the order write. A separate relay service reads committed outbox rows and sends them to Kafka. This ensures the Kafka message is produced if and only if the DB transaction committed. Second, the downstream consumer uses a content-hash idempotency key derived from the message content, so that even if the relay service delivers the outbox message more than once (which at-least-once delivery permits), the consumer deduplicates at the Stripe layer.

// OrderService.java — safe: transactional outbox instead of direct @KafkaClient send
@Singleton
public class OrderService {

    private final OrderRepository orderRepository;
    private final PaymentOutboxRepository paymentOutboxRepository;

    public OrderService(OrderRepository orderRepository,
                        PaymentOutboxRepository paymentOutboxRepository) {
        this.orderRepository = orderRepository;
        this.paymentOutboxRepository = paymentOutboxRepository;
    }

    @Retryable(attempts = "3", delay = "2s", includes = DataAccessException.class)
    @Transactional
    public void processOrder(String orderId, String customerId, int amountCents) {

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

        // Write order to DB
        Order order = new Order(orderId, customerId, amountCents, "PROCESSING");
        orderRepository.save(order);

        // Write outbox record to DB in the SAME @Transactional scope
        // If the transaction rolls back, both the order row AND the outbox row are rolled back
        // The outbox row is the "intent to send" — it only reaches Kafka after commit
        PaymentOutbox outbox = new PaymentOutbox();
        outbox.setOutboxKey(outboxKey);    // content-hash — deduplication key for consumer
        outbox.setCustomerId(customerId);
        outbox.setOrderId(orderId);
        outbox.setAmountCents(amountCents);
        outbox.setStatus("PENDING");
        paymentOutboxRepository.save(outbox);

        // Update order status
        order.setStatus("PAYMENT_QUEUED");
        orderRepository.update(order);
        // If this throws, the entire transaction rolls back — neither order nor outbox row committed
        // No Kafka message will be sent for this attempt
    }
}

// OutboxRelayService.java — reads committed outbox rows and sends to Kafka
@Singleton
public class OutboxRelayService {

    private final PaymentOutboxRepository paymentOutboxRepository;
    private final PaymentCommandClient paymentCommandClient;

    public OutboxRelayService(PaymentOutboxRepository paymentOutboxRepository,
                              PaymentCommandClient paymentCommandClient) {
        this.paymentOutboxRepository = paymentOutboxRepository;
        this.paymentCommandClient = paymentCommandClient;
    }

    @Scheduled(fixedDelay = "1s")
    @Transactional
    void relay() {
        List<PaymentOutbox> pending = paymentOutboxRepository.findByStatus("PENDING");
        for (PaymentOutbox outbox : pending) {
            // The PaymentCommand carries the content-hash key so the consumer can use it
            PaymentCommand cmd = new PaymentCommand(
                outbox.getCustomerId(),
                outbox.getOrderId(),
                outbox.getAmountCents(),
                outbox.getOutboxKey()        // idempotencyKey field in the message
            );
            paymentCommandClient.sendPaymentCommand(cmd);
            outbox.setStatus("RELAYED");
            paymentOutboxRepository.update(outbox);
        }
    }
}

// PaymentConsumer.java — safe: uses content-hash key from the message (set by the relay)
@Singleton
public class PaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

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

        // Content-hash key from the message — stable across every redelivery
        String idempotencyKey = command.idempotencyKey(); // set by OutboxRelayService

        // Check for prior processing — handles relay at-least-once delivery
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            return;  // idempotent skip
        }

        // PENDING row before Stripe call
        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);
        record.setCustomerId(command.customerId());
        record.setOrderId(command.orderId());
        record.setAmountCents(command.amountCents());
        record.setStatus("PENDING");
        paymentRecordRepository.saveAndFlush(record);

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

        record.setStripeChargeId(charge.chargeId());
        record.setStatus("COMPLETED");
        paymentRecordRepository.update(record);
    }
}

With this design, the only way a Kafka message reaches the payment-commands topic is if the corresponding PaymentOutbox row was committed to the database, which only happens if the surrounding @Transactional in processOrder committed. @Retryable on processOrder can safely fire: if the DB write rolled back (so neither the order row nor the outbox row committed), the relay will find no PENDING outbox rows for this order and no message will be sent. If the DB write committed, the relay will find the PENDING outbox row, send the message once, and mark the outbox row as RELAYED. Even if the relay service restarts mid-relay and sends the message twice (at-least-once), the consumer’s content-hash key and existsByIdempotencyKey check make the consumer idempotent.

Mode 3: @KafkaListener(batch=true) mid-batch failure — entire batch redelivered from the last committed offset — records already charged on Stripe in the first pass are re-processed with new UUID.randomUUID() values — ch_B for each

Micronaut Kafka’s batch=true flag on @KafkaListener changes the listener method signature from a per-record callback to a batch callback: the method receives a List<ConsumerRecord<K, V>> containing all records from the current consumer poll. Micronaut commits the consumer group offset after the batch method returns without throwing. If the batch method throws, the offset is not committed, and Kafka redelivers the entire batch on the next poll. This redelivery semantics interacts badly with UUID.randomUUID() idempotency keys: records that were processed (Stripe charged) in the first pass are redelivered with a new method invocation, and the per-record UUID.randomUUID() call generates a new key for each of them.

// BatchPaymentConsumer.java — unsafe: UUID.randomUUID() per record in batch listener
@Singleton
public class BatchPaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    public BatchPaymentConsumer(PaymentRecordRepository paymentRecordRepository,
                                StripeClient stripeClient) {
        this.paymentRecordRepository = paymentRecordRepository;
        this.stripeClient = stripeClient;
    }

    @KafkaListener(groupId = "batch-payment-processor", batch = true)
    @Topic("payment-commands")
    public void processBatch(List<ConsumerRecord<String, PaymentCommand>> records) {
        for (ConsumerRecord<String, PaymentCommand> record : records) {
            PaymentCommand command = record.value();

            // UUID.randomUUID() is called once per record on EACH batch invocation
            // If the batch is redelivered, this produces a new UUID for every record
            String idempotencyKey = UUID.randomUUID().toString();

            // Call Stripe for this record
            ChargeResult charge = stripeClient.createCharge(
                command.customerId(),
                command.amountCents(),
                idempotencyKey
            );

            // Save result to DB — may throw DataAccessException for some record N
            PaymentRecord paymentRecord = new PaymentRecord();
            paymentRecord.setIdempotencyKey(idempotencyKey);
            paymentRecord.setCustomerId(command.customerId());
            paymentRecord.setOrderId(command.orderId());
            paymentRecord.setStripeChargeId(charge.chargeId());
            paymentRecord.setStatus("COMPLETED");
            paymentRecordRepository.save(paymentRecord);
            // If save() throws for record N:
            //   - Records 1..N-1: Stripe charges AND DB saves both committed (individually)
            //   - Record N: Stripe charge committed, DB save NOT committed
            //   - Records N+1..end: not processed yet
            // The batch method throws. Micronaut Kafka does NOT commit the Kafka offset.
            // On next poll, Kafka redelivers the ENTIRE batch from the last committed offset.
            // Records 1..N-1 are re-processed with NEW UUID values — ch_B for each.
        }
    }
}

The developer’s mental model: each record in the batch is processed and saved independently. Records 1 through N-1 succeeded — their paymentRecordRepository.save() calls returned without error. Record N’s save failed. The developer expects that on redelivery, “Kafka knows” that records 1 through N-1 were handled and will redeliver only from record N onward.

This expectation is incorrect. The Kafka consumer group protocol tracks progress using partition offsets. A consumer commits an offset to indicate that all messages at that offset and earlier have been processed. Kafka does not have a per-message acknowledgment mechanism within a batch — offset granularity is per message, but offset commit happens per batch (in Micronaut Kafka’s default ackMode=AUTO for batch listeners). If the batch method throws before returning, no offset commit occurs for this batch. On the next poll, the Kafka broker delivers records starting at the last committed offset, which is before the first record of the failed batch.

The sequence of events for a batch of five records [R1, R2, R3, R4, R5] at offsets [50, 51, 52, 53, 54], where last committed offset is 49:

1. Micronaut Kafka polls the payment-commands topic. Returns [R1, R2, R3, R4, R5]. Invokes processBatch([R1..R5]).
2. R1: UUID.randomUUID() = key_1A. stripeClient.createCharge(cus_1, 9900, key_1A) → ch_1A committed on Stripe. paymentRecordRepository.save(record for R1) — succeeds. Row with key_1A and ch_1A committed to DB in auto-commit mode (no wrapping @Transactional on the batch method).
3. R2: UUID.randomUUID() = key_2A. Stripe → ch_2A committed. DB save — succeeds.
4. R3: UUID.randomUUID() = key_3A. Stripe → ch_3A committed. DB save — throws DataIntegrityViolationException (e.g., orderId of R3 is a duplicate due to an upstream producer bug that sent R3 twice in separate earlier batches).
5. The exception propagates out of the processBatch method. Micronaut Kafka catches it. Offset 54 is NOT committed. The consumer group’s committed offset remains at 49.
6. Micronaut Kafka pauses briefly and polls again. Kafka returns [R1, R2, R3, R4, R5] again (starting at offset 50, since the committed offset is 49).
7. processBatch([R1..R5]) is invoked again on the redelivered batch.
8. R1 (redelivery): UUID.randomUUID() = key_1B (statistically impossible to equal key_1A). stripeClient.createCharge(cus_1, 9900, key_1B) is called. Stripe checks its idempotency log: key_1B is not present (key_1A is, with ch_1A, but key_1B is distinct). Stripe creates ch_1B. Customer 1 is charged twice.
9. R2 (redelivery): key_2B generated. Stripe creates ch_2B. Customer 2 is charged twice.
10. R3 (redelivery): key_3A′ generated. Stripe creates ch_3A′. DB save may succeed or fail again depending on whether the duplicate condition persists.
11. R4, R5: processed for the first time. Single charges each.

The result: customers 1 and 2 are each double-charged. Customers 4 and 5 are correctly charged once. Customer 3’s outcome depends on whether the underlying DB issue was resolved.

Why the DB-level success of records 1 and 2 does not protect them from Kafka redelivery

A subtle and important distinction: even if the individual paymentRecordRepository.save() calls for records 1 and 2 used @Transactional at the per-record level (so that their DB writes committed independently before the batch method threw), those DB commits have no effect on the Kafka consumer group offset. Kafka offset commits and JDBC transaction commits are managed by completely separate systems. The DB correctly records that R1 and R2 were processed — which is useful for the fix — but Kafka does not consult the DB when deciding which offsets to redeliver. The broker’s only source of truth for “what has been processed” is the offset committed by the consumer group, which was 49 before the batch and remains 49 after the batch method threw.

This also means that if a developer adds @Transactional to the entire batch method (wrapping all records in a single JDBC transaction), the failure at R3 would roll back R1 and R2’s DB saves — but the Stripe charges for ch_1A and ch_2A are still committed on Stripe. On redelivery, the DB is clean for R1 and R2 (no prior DB record), so the consumer sees no evidence of prior processing and charges Stripe again with new UUIDs.

Fix: content-hash key per record from stable message fields + skip-on-prior-processing check

The fix applies the same content-hash pattern at the per-record level within the batch. Before calling Stripe for any record, the consumer checks whether a prior processing record exists in the DB for the content-hash key. If yes, the record is skipped. This check is safe to make even when the record is being processed for the first time, because the content-hash key is derived from the same message fields that the Kafka broker stores and redelivers unchanged.

// BatchPaymentConsumer.java — safe: content-hash per record + skip-on-prior-processing
@Singleton
public class BatchPaymentConsumer {

    private final PaymentRecordRepository paymentRecordRepository;
    private final StripeClient stripeClient;

    @KafkaListener(groupId = "batch-payment-processor", batch = true)
    @Topic("payment-commands")
    public void processBatch(List<ConsumerRecord<String, PaymentCommand>> records) {
        for (ConsumerRecord<String, PaymentCommand> kafkaRecord : records) {
            processRecord(kafkaRecord.value());
        }
        // Micronaut Kafka commits the offset for this batch after this method returns
    }

    @Transactional
    protected void processRecord(PaymentCommand command) {
        // Content-hash key — same value on every redelivery of this message
        String idempotencyKey = buildContentHash(
            command.customerId(),
            command.orderId(),
            String.valueOf(command.amountCents())
        );

        // Skip records already processed (handles: batch redelivery after mid-batch failure)
        if (paymentRecordRepository.existsByIdempotencyKey(idempotencyKey)) {
            // Already have a COMPLETED or PENDING record — idempotent skip
            return;
        }

        // PENDING row before Stripe call — serializes concurrent consumer instances
        PaymentRecord record = new PaymentRecord();
        record.setIdempotencyKey(idempotencyKey);
        record.setCustomerId(command.customerId());
        record.setOrderId(command.orderId());
        record.setAmountCents(command.amountCents());
        record.setStatus("PENDING");
        paymentRecordRepository.saveAndFlush(record);
        // If saveAndFlush throws DataIntegrityViolationException, another consumer instance
        // already inserted this PENDING row — this record is being handled concurrently.
        // Let the exception propagate — the batch will redeliver, and by then the
        // concurrent consumer will have committed (or failed, leaving a PENDING row to retry).

        // Stripe charge — content-hash key ensures Stripe deduplicates across redeliveries
        ChargeResult charge = stripeClient.createCharge(
            command.customerId(),
            command.amountCents(),
            idempotencyKey
        );

        // Update to COMPLETED
        record.setStripeChargeId(charge.chargeId());
        record.setStatus("COMPLETED");
        paymentRecordRepository.update(record);
    }

    private String buildContentHash(String... inputs) {
        String raw = String.join("|", inputs);
        try {
            MessageDigest sha256 = MessageDigest.getInstance("SHA-256");
            byte[] hash = sha256.digest(raw.getBytes(StandardCharsets.UTF_8));
            return "kb_" + HexFormat.of().formatHex(hash).substring(0, 40);
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("SHA-256 not available", e);
        }
    }
}

With this design, when the batch is redelivered after a mid-batch failure at R3:

• R1 (redelivery): content-hash key is the same as in the first pass (same customerId, orderId, amountCents). existsByIdempotencyKey() returns true (the COMPLETED row from the first pass is in the DB). Method returns without calling Stripe. No second charge.
• R2 (redelivery): same content-hash. COMPLETED row exists. Skip. No second charge.
• R3 (redelivery): content-hash is derived from R3’s fields. If the PENDING row from the first pass is in the DB (the saveAndFlush succeeded before the exception), existsByIdempotencyKey() returns true and Stripe is not called again (but ch_3A is already committed — a reconciliation job is needed to confirm the COMPLETED status for R3). If R3’s save failed with a UNIQUE violation due to an upstream bug, the content-hash key may be shared with another order, which surfaces as a COMPLETED row for that other order.
• R4, R5: no prior processing record. PENDING row inserted. Stripe called. COMPLETED.

An alternative to per-record @Transactional is to keep the entire processBatch method without @Transactional and let each per-record operation commit independently. The important invariant is that the PENDING row INSERT and the Stripe call happen within the same per-record scope: PENDING row committed to DB first, then Stripe called, then COMPLETED row update. This ordering ensures that a consumer failure after the PENDING INSERT but before the Stripe call leaves a PENDING row in the DB, which the redelivered record will find via existsByIdempotencyKey() and treat as in-progress (allowing the PENDING logic to verify via Stripe’s charge lookup whether ch was actually created).

Integration tests for all three modes

All three failure modes are reproducible in automated tests using Testcontainers (Apache Kafka, PostgreSQL 15), WireMock for Stripe stubs, and Micronaut’s @MicronautTest with embedded application context. The tests verify the failure in the unsafe code and the correct behavior in the fixed code.

// KafkaStripeIntegrationTest.java

@MicronautTest
@Testcontainers
class KafkaStripeIntegrationTest {

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

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15")
        .withDatabaseName("testdb").withUsername("test").withPassword("test");

    @RegisterExtension
    static WireMockExtension wireMock = WireMockExtension.newInstance()
        .options(wireMockConfig().dynamicPort()).build();

    @Inject
    PaymentConsumer paymentConsumer;           // unsafe version for modes 1 + 3

    @Inject
    SafePaymentConsumer safePaymentConsumer;  // fixed version

    @Inject
    BatchPaymentConsumer batchConsumer;       // unsafe batch for mode 3

    @Inject
    SafeBatchPaymentConsumer safeBatchConsumer;

    @Inject
    OrderService orderService;               // unsafe @KafkaClient version for mode 2

    @Inject
    SafeOrderService safeOrderService;       // transactional outbox version

    @Inject
    PaymentRecordRepository paymentRecordRepository;

    @Inject
    @KafkaProducer
    ProducerBean testProducer;               // for sending test messages to the topic

    // ---- Mode 1: @Retryable re-executes method body with new UUID ---- //

    @Test
    void mode1_retryableRegeneratesUUID_doubleChargesStripe() throws Exception {
        // Arrange: Stripe stubs — two distinct charge IDs for the first two calls
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .inScenario("mode1")
            .whenScenarioStateIs(STARTED)
            .willReturn(okJson("{\"id\":\"ch_A\",\"object\":\"charge\",\"status\":\"succeeded\"}"))
            .willSetStateTo("attempt2"));
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .inScenario("mode1")
            .whenScenarioStateIs("attempt2")
            .willReturn(okJson("{\"id\":\"ch_B\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        // Arrange: cause the first @Retryable attempt's DB save to fail
        simulateDbSaveFailureOnFirstAttempt();

        PaymentCommand command = new PaymentCommand("cus_1", "order_1", 4999);

        // Act: invoke the unsafe consumer directly (bypassing Kafka for determinism)
        try {
            paymentConsumer.onPaymentCommand(command);  // @Retryable fires internally
        } catch (Exception e) {
            // Swallowed after @Retryable exhaustion
        }

        // Assert: Stripe received two requests with different Idempotency-Key header values
        List<LoggedRequest> stripeRequests = wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")));
        assertThat(stripeRequests).hasSize(2);
        String key1 = extractIdempotencyKey(stripeRequests.get(0));
        String key2 = extractIdempotencyKey(stripeRequests.get(1));
        assertThat(key1).isNotEqualTo(key2);   // two distinct keys = double charge
    }

    @Test
    void mode1_fixed_contentHashKey_stripeDeduplicates() throws Exception {
        // Arrange: Stripe returns ch_A for the content-hash key on every call
        String expectedKey = buildContentHash("cus_1", "order_1", "4999");
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .withHeader("Idempotency-Key", equalTo(expectedKey))
            .willReturn(okJson("{\"id\":\"ch_A\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        simulateDbSaveFailureOnFirstAttempt();

        PaymentCommand command = new PaymentCommand("cus_1", "order_1", 4999);

        // Act: invoke the safe consumer
        safePaymentConsumer.onPaymentCommand(command);

        // Assert: all Stripe requests used the same content-hash key
        List<LoggedRequest> reqs = wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")));
        assertThat(reqs).allSatisfy(req ->
            assertThat(extractIdempotencyKey(req)).isEqualTo(expectedKey));
    }

    // ---- Mode 2: @KafkaClient non-transactional + consumer UUID ---- //

    @Test
    void mode2_kafkaClientInsideTransactional_duplicateMessageDoubleCharges() throws Exception {
        // Arrange: consumer Stripe stubs — different ch_ID per call (UUID keys differ)
        AtomicInteger stripeCallCount = new AtomicInteger(0);
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .willReturn(aResponse()
                .withStatus(200)
                .withBody("{\"id\":\"ch_" + stripeCallCount.incrementAndGet() + "\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        // Simulate the order update failing on first attempt
        simulateOrderUpdateFailureOnFirstAttempt();

        // Act: call unsafe OrderService (which sends @KafkaClient inside @Transactional)
        try {
            orderService.processOrder("ORDER-1", "cus_1", 9900);
        } catch (Exception e) {
            // @Retryable succeeded on second attempt
        }

        // Wait for Kafka consumer to process both messages
        await().atMost(10, SECONDS).until(() ->
            wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges"))).size() >= 2);

        // Assert: Stripe received two calls with different idempotency keys
        List<LoggedRequest> reqs = wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")));
        assertThat(reqs).hasSize(2);
        assertThat(extractIdempotencyKey(reqs.get(0)))
            .isNotEqualTo(extractIdempotencyKey(reqs.get(1)));

        // Also assert: there are two messages on the payment-commands topic
        long topicMessageCount = countMessagesOnTopic(kafka.getBootstrapServers(), "payment-commands");
        assertThat(topicMessageCount).isEqualTo(2);  // cmd1 from failed attempt + cmd2 from success
    }

    @Test
    void mode2_fixed_outboxPattern_singleStripeCharge() throws Exception {
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .willReturn(okJson("{\"id\":\"ch_A\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        simulateOrderUpdateFailureOnFirstAttempt();

        safeOrderService.processOrder("ORDER-1", "cus_1", 9900);

        await().atMost(10, SECONDS).until(() ->
            wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges"))).size() >= 1);

        // Relay sends exactly one message regardless of @Retryable attempts
        assertThat(wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")))).hasSize(1);
    }

    // ---- Mode 3: batch=true redelivery double-charges already-processed records ---- //

    @Test
    void mode3_batchRedelivery_doubleChargesProcessedRecords() throws Exception {
        // Arrange: 5 records. Stripe succeeds for all. DB save fails on record 3 (first pass).
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .willReturn(okJson("{\"id\":\"ch_DYNAMIC\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        List<ConsumerRecord<String, PaymentCommand>> batch = List.of(
            makeRecord("cus_1", "order_1", 1000),
            makeRecord("cus_2", "order_2", 2000),
            makeRecord("cus_3", "order_3", 3000), // DB save will fail for this one
            makeRecord("cus_4", "order_4", 4000),
            makeRecord("cus_5", "order_5", 5000)
        );

        simulateDbSaveFailureForRecord("order_3");

        // Act: first batch pass — method throws after processing records 1 and 2
        try {
            batchConsumer.processBatch(batch);
        } catch (Exception ignored) {}

        // Kafka redelivers. Second pass — records 1 and 2 are re-processed.
        resetDbSaveFailure();
        batchConsumer.processBatch(batch);

        // Assert: Stripe received charges for cus_1 and cus_2 twice each (ch_A and ch_B)
        List<LoggedRequest> reqs = wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")));

        Map<String, Long> keyFrequency = reqs.stream()
            .collect(Collectors.groupingBy(
                req -> extractCustomerFromBody(req.getBodyAsString()),
                Collectors.counting()
            ));

        assertThat(keyFrequency.get("cus_1")).isEqualTo(2L);   // double charged
        assertThat(keyFrequency.get("cus_2")).isEqualTo(2L);   // double charged
    }

    @Test
    void mode3_fixed_contentHashKey_batchRedeliveryIdempotent() throws Exception {
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .willReturn(okJson("{\"id\":\"ch_DYNAMIC\",\"object\":\"charge\",\"status\":\"succeeded\"}")));

        List<ConsumerRecord<String, PaymentCommand>> batch = List.of(
            makeRecord("cus_1", "order_1", 1000),
            makeRecord("cus_2", "order_2", 2000),
            makeRecord("cus_3", "order_3", 3000),
            makeRecord("cus_4", "order_4", 4000),
            makeRecord("cus_5", "order_5", 5000)
        );

        simulateDbSaveFailureForRecord("order_3");

        try {
            safeBatchConsumer.processBatch(batch);
        } catch (Exception ignored) {}

        resetDbSaveFailure();
        safeBatchConsumer.processBatch(batch);  // redelivery

        // Assert: Stripe received exactly one charge per customer (content-hash deduplication)
        List<LoggedRequest> reqs = wireMock.findAll(postRequestedFor(urlEqualTo("/v1/charges")));
        Map<String, Long> keyFrequency = reqs.stream()
            .collect(Collectors.groupingBy(
                req -> extractCustomerFromBody(req.getBodyAsString()),
                Collectors.counting()
            ));

        assertThat(keyFrequency.getOrDefault("cus_1", 0L)).isLessThanOrEqualTo(1L);
        assertThat(keyFrequency.getOrDefault("cus_2", 0L)).isLessThanOrEqualTo(1L);
    }
}

Comparison table: three Micronaut Kafka + Stripe failure modes

What Micronaut Kafka is and is not

Micronaut Kafka is an at-least-once messaging layer. @KafkaClient produces messages with at-least-once delivery semantics when configured with enable.idempotence=false (the default): a message may be delivered to the broker more than once if the producer retries after a network timeout. @KafkaListener consumes messages with at-least-once delivery semantics: the same message may be delivered to the consumer more than once if the consumer restarts, rebalances, or fails to commit an offset.

At-least-once delivery is the correct default for most use cases. It is preferable to at-most-once delivery (which would lose messages) and simpler to implement than exactly-once delivery (which requires Kafka transactions, enable.idempotence=true on the producer, and isolation.level=read_committed on the consumer). The implication for Stripe integration is that every component that calls Stripe must be designed for idempotent re-execution: the Stripe call must produce the same observable effect (the same charge committed) on every invocation for the same logical billing event, whether the invocation is the first, a @Retryable retry, a Kafka redelivery, or a batch re-processing pass.

The three failure modes in this post are each a violation of the idempotency requirement at a different layer: the AOP retry layer (Mode 1), the message production layer (Mode 2), and the batch consumer offset-commit layer (Mode 3). The fix in all three cases is the same at its core: derive the Stripe idempotency key from stable, content-addressable inputs rather than from UUID.randomUUID(), which generates a new value on every method invocation regardless of whether the method is executing for the first time or the fifth. The secondary fix — the PENDING row INSERT before the Stripe call — serializes concurrent consumer instances that might receive the same message due to Kafka partition reassignment, and leaves a durable marker for reconciliation when the consumer fails between the Stripe call and the DB update.

Keybrake and Micronaut Kafka Stripe integrations

The three failure modes in this post share a detection problem. In Mode 1, ch_A is committed on Stripe with no DB record — visible only in the Stripe dashboard, invisible in the application’s payment_records table. In Mode 2, two messages on the Kafka topic look identical from the broker’s perspective (same payload, different offsets); the double charge is detectable only by correlating the consumer’s two DB rows with the Stripe dashboard. In Mode 3, the double charges for records 1–N-1 are each recorded in the DB (two rows per customer, two distinct idempotency keys), but reconciling these with Stripe requires querying by customerId and amount window rather than by idempotency key, since the second-pass key differs from the first-pass key.

In production, all three failure modes are difficult to catch in real time. The application does not throw an unhandled exception visible in monitoring (or if it does, the exception is the DB write failure, not the double charge). Stripe returns HTTP 200 for both ch_A and ch_B. The customer sees two charges on their statement. The support ticket arrives days later.

Keybrake operates as a reverse proxy between the Micronaut application and Stripe. Every charge request is recorded in Keybrake’s audit log with the Idempotency-Key header value, the customer ID, and the response charge ID. When a consumer fires twice for the same logical event and each call uses a different UUID key, Keybrake records both: two charge events for the same customer and amount within a short window, each with a distinct idempotency key, neither referencing the other. This pattern is the earliest detectable signal for all three failure modes — available immediately after the second charge is committed, before any reconciliation job, and before the customer notices. Keybrake’s per-vendor daily spend cap provides a second line of defense: a buggy batch consumer that double-charges every record in a large batch will hit the spend cap and halt further charges before the damage scales.