Keybrake

Spring Cloud Stream Function Stripe Integration: How SCS Function Composition Retry Re-Executes processStripeCharge After ch_A Commits Generating UUID_B Committing ch_B, How Consumer<Flux<BillingEvent>> flatMap Processes a Redelivered Billing Message Concurrently With the Original Generating UUID_B, and How @PollableBean Supplier Re-Polls an Unprocessed Outbox Row Before the Consumer Marks It Complete Committing ch_B

By Keybrake —

The Spring Cloud Stream Function programming model — the @Bean Consumer<T>, Function<T,R>, and Supplier<T> paradigm introduced in SCS 3.x — has three failure modes that produce duplicate Stripe charges, and none of them come from binder-level Kafka, RabbitMQ, or Pulsar semantics. They come from abstractions that the functional programming model adds on top of the binder: function composition chains, reactive Flux pipelines, and polled supplier scheduling. If you have migrated from @StreamListener to the functional model and your billing consumers do not use a stable idempotency key, any of these three scenarios can silently commit ch_B.

This post covers all three end to end: the exact SCS Function model configuration that enables each failure, the step-by-step timeline showing when ch_A commits and when UUID_B is generated, the specific misconception each one exploits, and the idempotency fix that makes your billing consumer safe regardless of how the SCS Function runtime delivers or retries the message. Code samples are Spring Cloud Stream 4.x, Spring Cloud Function 4.x, Spring Boot 3.x, WireMock, Testcontainers, and JUnit 5.

Background: what the SCS Function programming model adds on top of the binder

The SCS functional programming model is a layer on top of the binder infrastructure. The binder (Kafka, RabbitMQ, Pulsar) handles message transport — consumer group assignment, offset tracking, redelivery, and acknowledgement. The functional model handles message routing at the application layer: it maps incoming binder messages to @Bean functions, manages function composition and chaining, wraps reactive functions in Flux pipelines, and optionally drives a supplier-based polling loop on a configurable schedule.

Each of the three failure modes in this post is a direct consequence of one of these application-layer responsibilities.

Function composition is a single atomic consumer invocation. When you compose two functions with spring.cloud.function.definition=enrichBillingEvent|processStripeCharge, the SCS runtime creates a single composed Function<In, Out> that runs enrichBillingEvent then processStripeCharge in sequence. This composed function is registered as a single consumer binding. When the binder retries a message — because the composed function threw an exception — the binder re-invokes the entire composed function from the start, not from the point of failure. There is no per-function state carried between retry attempts. enrichBillingEvent runs again, then processStripeCharge runs again, calling UUID.randomUUID() again and producing a different idempotency key.

Reactive consumer functions use Project Reactor operators that allow concurrent execution. When you write a Consumer<Flux<BillingEvent>> and use .flatMap(event -> chargeStripe(event)) to process each event, Project Reactor's flatMap operator subscribes to all inner publishers concurrently by default. If the binder delivers the same message twice into the Flux pipeline — because a Kafka rebalance occurred and the offset was not yet committed — flatMap will execute both concurrently. Each concurrent invocation calls UUID.randomUUID() independently and produces a different idempotency key.

Polled suppliers run on a fixed schedule independent of downstream consumer state. When you annotate a Supplier<List<BillingEvent>> with @PollableBean and configure spring.cloud.stream.poller.fixed-delay=5000, the SCS runtime schedules the supplier to fire every 5 seconds regardless of whether the downstream consumer has finished processing the previous batch. If the supplier queries a DB outbox table for processed_at IS NULL rows, and the downstream consumer takes longer than 5 seconds to mark a row as processed (because a Stripe 3DS call takes 12 seconds), the same row will be polled twice. Both emissions reach the consumer, both call UUID.randomUUID(), and both commit to Stripe.

With that context in place, the three failure modes follow directly.

Failure mode 1: spring.cloud.function.definition=enrichBillingEvent|processStripeCharge — retry re-executes the full composed function re-invoking processStripeCharge with a new UUID.randomUUID() after ch_A was already committed

The setup

A team runs a subscription billing service backed by Apache Kafka. The billing pipeline is a Spring Cloud Stream 4.x application using the SCS Kafka Binder. The team recently migrated from @StreamListener to the SCS functional programming model following the Spring Boot 3.x upgrade.

The developer decomposes the billing consumer into two separate functions to improve testability. enrichBillingEvent is a pure function: it reads customer metadata from a Redis cache and attaches the Stripe customer ID and tax rate to the BillingEvent. processStripeCharge takes the enriched event, calls the Stripe Charges API, and writes a BillingRecord to PostgreSQL. The developer composes them with spring.cloud.function.definition=enrichBillingEvent|processStripeCharge.

# application.yml
spring:
  cloud:
    function:
      definition: enrichBillingEvent|processStripeCharge
    stream:
      bindings:
        # SCS auto-derives binding name from composed function definition
        enrichBillingEvent-processStripeCharge-in-0:
          destination: billing
          group: billing-group
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000
            back-off-max-interval: 10000
            back-off-multiplier: 2.0
      kafka:
        bindings:
          enrichBillingEvent-processStripeCharge-in-0:
            consumer:
              enable-dlq: true
              dlq-name: billing.DLT

The two bean functions are defined in separate configuration classes:

// Pure enrichment function — no side effects
@Configuration
public class BillingEnricher {

    private final CustomerMetadataCache metadataCache;

    @Bean
    public Function<BillingEvent, BillingEvent> enrichBillingEvent() {
        return event -> {
            CustomerMetadata meta = metadataCache.get(event.getCustomerId());
            return event.toBuilder()
                .stripeCustomerId(meta.getStripeCustomerId())
                .taxRateBps(meta.getTaxRateBps())
                .build();
        };
    }
}

// Stripe charge function — external call + DB write
@Configuration
public class BillingCharger {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<BillingEvent> processStripeCharge() {
        return event -> {

            // BUG: fresh UUID per invocation — no idempotency key
            String idempotencyKey = UUID.randomUUID().toString();

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Subscription " + event.getSubscriptionId())
                .build();

            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            billingRecordRepo.save(new BillingRecord(
                event.getSubscriptionId(),
                event.getBillingPeriod(),
                charge.getId(),
                idempotencyKey
            ));
        };
    }
}

The failure timeline

A billing run processes 600 subscription orders. Message M1 carries a BillingEvent for customer C1, a US business customer paying with a corporate card that triggers Stripe Radar review for high-value amounts. The charge is $1,200.

The DLT entry records DataIntegrityViolationException in its headers. The on-call engineer reading the DLT header sees a database error and will likely interpret this as a failure before Stripe was called — compounding the investigation difficulty on top of the triple charge already committed.

The misconception

"SCS Function composition retries are scoped to the function that threw the exception. enrichBillingEvent completed successfully before the exception in processStripeCharge — there is no reason for the SCS runtime to re-invoke enrichBillingEvent on retry. The composition is a pipeline, and the retry resumes from the failed stage."

This is wrong in a specific and important way. Spring Cloud Stream's consumer binding retry is implemented at the binder message listener level, not at the function composition level. The binder receives a raw Kafka ConsumerRecord and invokes the registered message handler — which is the composed Function<In, Out> as a single unit. When the message handler throws, the binder's retry mechanism re-invokes the message handler with the same ConsumerRecord. The binder has no knowledge of the internal stages of the composed function; it only knows about the single message handler it registered.

This means that every retry re-executes the entire composed function from the beginning. enrichBillingEvent runs again on every retry attempt. processStripeCharge runs again on every retry attempt. Each invocation of processStripeCharge calls UUID.randomUUID() independently, producing a different idempotency key and creating a new Stripe charge.

The analogous behavior in a non-composed single-function consumer is identical and easier to reason about — but the composition pattern makes it much less obvious because developers naturally think of composed functions as a pipeline where each stage has its own state.

Diagnosing this failure in production

The signature of this failure mode in Stripe's dashboard is a cluster of charges for the same customer with the same amount committed within a 10-30 second window, each under a different charge ID and a different idempotency key. The idempotency keys will all be UUID v4 strings with no detectable relationship to each other or to the subscription ID.

In your application logs, you will see the initial DataIntegrityViolationException followed by SCS retry log lines (Retrying failed message), followed by additional DataIntegrityViolationException entries. The retry log will show the same Kafka offset being retried, confirming it is the same message. What the logs will not show is that Stripe was called successfully on every attempt before the exception — because the exception comes from the DB write after the Stripe commit, not from the Stripe call itself.

The fastest diagnostic query is to search Stripe for charges with your subscription ID in the description field within a short time window:

-- Check for duplicate Stripe charges for the same subscription and billing period
SELECT stripe_charge_id, idempotency_key, created_at
FROM billing_records
WHERE subscription_id = 'sub_C1'
  AND billing_period = '2026-10'
ORDER BY created_at ASC;

-- If this returns zero rows: DataIntegrityViolationException on ALL attempts —
-- the billing_records table has a stale row from a prior aborted run.
-- Check Stripe directly:
-- stripe charges list --customer cus_A1B2 --created[gte]=1729123200 --created[lte]=1729123400

The fix

The fix requires two changes. First, derive the Stripe idempotency key from stable fields of the BillingEvent payload — fields that are identical on every invocation for the same logical billing event — so that every retry attempt generates the same key and Stripe deduplicates the charge. Second, add a guard-before-Stripe check: query BillingRecord by the stable key before calling Stripe so that a retry that encounters a stale duplicate row in the DB skips the Stripe call entirely.

The stable key should encode the fields that uniquely identify a billing event: subscription ID, billing period, and amount in cents. A SHA-256 hex digest of these fields produces a 64-character deterministic string that is safe to use as a Stripe idempotency key (Stripe accepts idempotency keys up to 255 characters).

@Configuration
public class BillingCharger {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<BillingEvent> processStripeCharge() {
        return event -> {

            // Stable key — same value on every retry for the same billing event
            String idempotencyKey = sha256(
                event.getSubscriptionId() + "|" +
                event.getBillingPeriod() + "|" +
                event.getAmountCents()
            );

            // Guard: skip Stripe entirely if we already committed this charge
            if (billingRecordRepo.existsByIdempotencyKey(idempotencyKey)) {
                return; // ack — idempotent skip
            }

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Subscription " + event.getSubscriptionId())
                .build();

            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            billingRecordRepo.save(new BillingRecord(
                event.getSubscriptionId(),
                event.getBillingPeriod(),
                charge.getId(),
                idempotencyKey
            ));
        };
    }

    private static String sha256(String input) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }
}

With the stable key in place, the retry timeline changes completely. On retry 2, processStripeCharge computes the same idempotencyKey as on attempt 1. The guard check queries BillingRecord — but the record doesn't exist (the DataIntegrityViolationException prevented it from being saved). So the guard does not short-circuit. processStripeCharge calls Stripe with the same idempotency key. Stripe returns ch_A from its idempotency cache. No second charge is committed. The DB write on attempt 2 may need to resolve the stale duplicate row separately, but the charge deduplication is guaranteed regardless.

Test: Stripe called once despite function composition retry

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"billing"})
class FunctionCompositionRetryTest {

    @Autowired
    private KafkaTemplate<String, BillingEvent> kafkaTemplate;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9090));
        wireMock.start();
        // First Stripe call succeeds
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .inScenario("billing-retry")
            .whenScenarioStateIs(STARTED)
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":120000,\"status\":\"succeeded\"}")
            )
            .willSetStateTo("charged"));
        // Second Stripe call should return ch_A from idempotency cache
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .inScenario("billing-retry")
            .whenScenarioStateIs("charged")
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":120000,\"status\":\"succeeded\"}")
            ));
    }

    @Test
    void composedFunctionRetry_stripeCalledOnce_idempotentKey() throws Exception {
        BillingEvent event = BillingEvent.builder()
            .subscriptionId("sub_retry_test")
            .billingPeriod("2026-10")
            .amountCents(120000L)
            .stripeCustomerId("cus_test")
            .build();

        // Plant stale duplicate row to trigger DataIntegrityViolationException on first save
        billingRecordRepo.save(new BillingRecord("sub_retry_test", "2026-10",
            "ch_old", sha256("sub_retry_test|2026-10|120000")));

        kafkaTemplate.send("billing", event).get();

        // Wait for DLT routing (maxAttempts=3 exhausted)
        Thread.sleep(15_000);

        // Stripe must have been called — but only one distinct idempotency key used
        List<LoggedRequest> requests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(requests).hasSizeGreaterThanOrEqualTo(1);

        // All Stripe calls must use the same stable idempotency key
        Set<String> keys = requests.stream()
            .map(r -> r.getHeader("Idempotency-Key"))
            .collect(Collectors.toSet());
        assertThat(keys).hasSize(1);

        // No second charge committed — only ch_A in billing records
        List<BillingRecord> records = billingRecordRepo.findBySubscriptionIdAndBillingPeriod(
            "sub_retry_test", "2026-10");
        long distinctChargeIds = records.stream()
            .map(BillingRecord::getStripeChargeId)
            .distinct().count();
        assertThat(distinctChargeIds).isEqualTo(1);
    }

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

Failure mode 2: Consumer<Flux<BillingEvent>> with .flatMap(event -> chargeStripe(event)) — Kafka partition rebalance delivers the same message twice into the Flux pipeline; flatMap processes both concurrently; each invocation calls UUID.randomUUID() independently; Stripe commits ch_A and ch_B

The setup

A team runs a high-throughput billing service processing 50,000 subscription renewals per hour. To avoid blocking Kafka polling threads on Stripe HTTP calls, the developer implements the billing consumer as a reactive SCS function using Consumer<Flux<BillingEvent>>. Inside the consumer, each event is mapped to a Mono<Charge> via a non-blocking reactive Stripe HTTP call and then flat-mapped to the billing record save operation. The developer uses .flatMap() to allow multiple Stripe calls to be in-flight simultaneously, increasing billing throughput.

# application.yml
spring:
  cloud:
    function:
      definition: processPayments
    stream:
      bindings:
        processPayments-in-0:
          destination: billing
          group: billing-group
      kafka:
        bindings:
          processPayments-in-0:
            consumer:
              start-offset: latest
              ack-mode: MANUAL_IMMEDIATE
@Configuration
public class ReactiveBillingConsumer {

    private final ReactiveStripeClient stripe;
    private final ReactiveBillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<Flux<BillingEvent>> processPayments() {
        return payments -> payments
            // BUG: flatMap allows unlimited concurrent subscriptions
            // — same message from a rebalance redelivery is processed in parallel
            .flatMap(event -> chargeStripe(event))
            .subscribe();
    }

    private Mono<Void> chargeStripe(BillingEvent event) {
        // BUG: UUID.randomUUID() called per-invocation — not per-billing-event
        String idempotencyKey = UUID.randomUUID().toString();

        return Mono.fromCallable(() ->
                ChargeCreateParams.builder()
                    .setAmount(event.getAmountCents())
                    .setCurrency("usd")
                    .setCustomer(event.getStripeCustomerId())
                    .setDescription("Subscription " + event.getSubscriptionId())
                    .build())
            .flatMap(params -> Mono.fromFuture(
                stripe.charges().createAsync(params,
                    RequestOptions.builder()
                        .setIdempotencyKey(idempotencyKey)
                        .build())))
            .flatMap(charge ->
                billingRecordRepo.save(new BillingRecord(
                    event.getSubscriptionId(),
                    event.getBillingPeriod(),
                    charge.getId(),
                    idempotencyKey)))
            .then();
    }
}

The failure timeline

The billing service runs two consumer pods. A rolling deployment starts at 14:03 UTC. The Kubernetes rollout terminates the old pod and starts the new one, which triggers a Kafka consumer group rebalance. Message M1, a BillingEvent for customer C2 with an amount of $299, was in-flight during the rebalance.

Customer C2 is charged $299 twice. Stripe has ch_A and ch_B as independent charges with no link between them. The DataIntegrityViolationException on Pod-B's save is logged as an error in the reactive pipeline, but by the time it is logged, the duplicate charge exists in Stripe.

The misconception

"Kafka at-least-once delivery means the same record is delivered at most once at a time. If a consumer is currently processing a record, Kafka will not redeliver it to another consumer until the offset is committed or the processing consumer is revoked and the record's offset is picked up by the new assignee. In the reactive pipeline, flatMap processes each delivered record once — it doesn't create duplicate subscriptions for the same record."

This misconception has two parts, both wrong in specific ways.

The first part conflates "a record is not delivered twice concurrently to the same consumer group" with "a record is not in-flight on two threads at the same time across the group." Kafka's consumer group protocol ensures that at most one consumer instance holds the partition assignment at any time. But when a rebalance transfers partition P0 from Pod-A to Pod-B, Pod-B starts consuming from the last committed offset — which may be behind the last processed offset if the reactive consumer uses ack-mode: MANUAL_IMMEDIATE and the offset commit has not yet been issued. Any record between the last committed offset and the rebalance point will be redelivered to Pod-B. If Pod-A's in-flight processing has not yet committed the offset, that record is delivered to both pods — sequentially from Kafka's perspective, but concurrently from the application's wall-clock perspective because Pod-A's Mono is still executing when Pod-B starts the same Mono.

The second part misunderstands flatMap's concurrency semantics. Project Reactor's flatMap operator eagerly subscribes to all inner publishers as they arrive, up to a default concurrency of Integer.MAX_VALUE. It does not wait for the first subscription to complete before subscribing to the second. If Pod-B's Flux receives M1 while Pod-A's Mono for M1 is still executing, flatMap on Pod-B subscribes immediately and both Monos execute concurrently — each calling UUID.randomUUID() independently.

The fix

Two changes are required. First, replace flatMap with concatMap for cases where ordering or single-flight semantics matter, or set an explicit concurrency limit on flatMap to a value that matches your use case. However, even with concatMap, the cross-pod rebalance redelivery is not prevented — it just means Pod-B processes M1 sequentially rather than Pod-A and Pod-B overlapping. The only thing that prevents the double charge in the cross-pod scenario is a stable idempotency key.

Second, move UUID.randomUUID() to outside the reactive pipeline and derive the key from stable event fields instead:

@Configuration
public class ReactiveBillingConsumer {

    private final ReactiveStripeClient stripe;
    private final ReactiveBillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<Flux<BillingEvent>> processPayments() {
        return payments -> payments
            // Use concatMap for ordering; stable key for idempotency
            .concatMap(event -> chargeStripe(event))
            .subscribe();
    }

    private Mono<Void> chargeStripe(BillingEvent event) {
        // Stable key — same value for the same billing event regardless of which pod processes it
        String idempotencyKey = sha256(
            event.getSubscriptionId() + "|" +
            event.getBillingPeriod() + "|" +
            event.getAmountCents()
        );

        return billingRecordRepo.existsByIdempotencyKey(idempotencyKey)
            .flatMap(exists -> {
                if (exists) {
                    return Mono.empty(); // already charged — skip Stripe
                }
                return Mono.fromCallable(() ->
                        ChargeCreateParams.builder()
                            .setAmount(event.getAmountCents())
                            .setCurrency("usd")
                            .setCustomer(event.getStripeCustomerId())
                            .setDescription("Subscription " + event.getSubscriptionId())
                            .build())
                    .flatMap(params -> Mono.fromFuture(
                        stripe.charges().createAsync(params,
                            RequestOptions.builder()
                                .setIdempotencyKey(idempotencyKey)
                                .build())))
                    .flatMap(charge ->
                        billingRecordRepo.save(new BillingRecord(
                            event.getSubscriptionId(),
                            event.getBillingPeriod(),
                            charge.getId(),
                            idempotencyKey)));
            })
            .then();
    }

    private static String sha256(String input) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }
}

With the stable key, even if Pod-A and Pod-B both process M1 with overlapping Stripe calls, Stripe will deduplicate: the second call with the same idempotency key will return ch_A from the cache. The guard-before-Stripe check adds an additional safety layer for cases where the BillingRecord was successfully saved by Pod-A before Pod-B executes its Stripe call — Pod-B will find the existing record and skip the Stripe call entirely without needing to rely on Stripe's idempotency cache.

SCS reactive consumer concurrency reference table

Operator Default concurrency Same-message parallel risk on rebalance Stable key needed?
flatMap(f) Unlimited (Integer.MAX_VALUE) Yes — within a pod if buffer > 1 Yes — essential
flatMap(f, 1) 1 (sequential within pod) Yes — cross-pod rebalance still possible Yes — essential
concatMap(f) 1 (sequential, no eager subscribe) Yes — cross-pod rebalance still possible Yes — essential
flatMapSequential(f) Unlimited in-flight, sequential emit Yes — within and cross-pod Yes — essential

No Reactor operator prevents cross-pod redelivery from a Kafka rebalance. Stable idempotency keys are the only mechanism that makes reactive billing consumers safe across rolling deployments, pod restarts, and partition reassignments.

Test: Stripe called once despite concurrent redelivery into the Flux pipeline

@SpringBootTest
@EmbeddedKafka(partitions = 1, topics = {"billing"})
class ReactiveFlatMapConcurrencyTest {

    @Autowired
    private KafkaTemplate<String, BillingEvent> kafkaTemplate;

    @Autowired
    private ReactiveBillingRecordRepository billingRecordRepo;

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9091));
        wireMock.start();
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withFixedDelay(200) // 200ms Stripe response — allows concurrency to manifest
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":29900,\"status\":\"succeeded\"}")));
    }

    @Test
    void reactiveFlatMap_concurrentRedelivery_stripeCalledWithSameKey() throws Exception {
        BillingEvent event = BillingEvent.builder()
            .subscriptionId("sub_reactive_test")
            .billingPeriod("2026-10")
            .amountCents(29900L)
            .stripeCustomerId("cus_test")
            .build();

        // Simulate rebalance redelivery: send the same event twice in rapid succession
        kafkaTemplate.send("billing", event).get();
        kafkaTemplate.send("billing", event).get();

        Thread.sleep(5_000);

        // Stripe may be called twice (once per delivery) but must use the same idempotency key
        List<LoggedRequest> requests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(requests.size()).isGreaterThanOrEqualTo(1);

        Set<String> idempotencyKeys = requests.stream()
            .map(r -> r.getHeader("Idempotency-Key"))
            .collect(Collectors.toSet());

        // All calls must use the same stable idempotency key
        assertThat(idempotencyKeys).hasSize(1);

        // Only one billing record saved
        List<BillingRecord> records = billingRecordRepo
            .findBySubscriptionIdAndBillingPeriod("sub_reactive_test", "2026-10")
            .collectList()
            .block();
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getStripeChargeId()).isEqualTo("ch_A");
    }

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

Failure mode 3: @PollableBean Supplier<List<BillingEvent>> with spring.cloud.stream.poller.fixed-delay=5000 — DB outbox row re-polled while the downstream Consumer's Stripe call is still in-flight; Consumer processes the same row twice; UUID.randomUUID() called independently on each processing; Stripe commits ch_B

The setup

A team implements the transactional outbox pattern for their billing service. Rather than publishing billing events directly from the transaction that creates the subscription, they write BillingEvent rows to an outbox table within the subscription creation transaction. A separate Spring Cloud Stream application polls the outbox table and feeds events into the billing pipeline.

The developer uses the SCS @PollableBean pattern to implement the outbox poller as a Supplier<List<BillingEvent>>. The supplier queries SELECT * FROM billing_outbox WHERE processed_at IS NULL ORDER BY created_at ASC LIMIT 10 and returns the result. The downstream consumer calls Stripe and then marks the row as processed by setting processed_at = NOW().

# application.yml
spring:
  cloud:
    function:
      definition: outboxPoller|processStripeCharge
    stream:
      bindings:
        # Supplier output binding
        outboxPoller-out-0:
          destination: billing-internal
        # Consumer input binding
        processStripeCharge-in-0:
          destination: billing-internal
          group: billing-internal-group
      poller:
        # BUG: 5-second poll interval — Stripe 3DS takes up to 15 seconds
        fixed-delay: 5000
        max-messages-per-poll: 10
@Configuration
public class OutboxPoller {

    private final BillingOutboxRepository outboxRepo;

    @Bean
    @PollableBean
    public Supplier<List<BillingEvent>> outboxPoller() {
        return () -> outboxRepo.findUnprocessed(10);
    }
}

@Configuration
public class BillingCharger {

    private final StripeClient stripe;
    private final BillingOutboxRepository outboxRepo;

    @Bean
    public Consumer<BillingEvent> processStripeCharge() {
        return event -> {

            // BUG: fresh UUID per invocation — no stable key
            String idempotencyKey = UUID.randomUUID().toString();

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Subscription " + event.getSubscriptionId())
                .build();

            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            // Mark the outbox row as processed after Stripe commit
            outboxRepo.markProcessed(event.getOutboxId(), charge.getId());
        };
    }
}

The failure timeline

The outbox table has 15 rows pending from the nightly billing run. The poller and consumer are both active. Among the pending rows is row R1: a BillingEvent for customer C3 with a $499 charge. C3 is a French customer paying with a Carte Bancaire card that requires 3D Secure authentication. Stripe's p99 latency for 3DS-enrolled French cards is approximately 12 seconds.

Customer C3 is charged $499 twice. If the second markProcessed overwrites the first, the outbox row records ch_A as the charge ID — making ch_B invisible to the application's own records. The only place ch_B appears is in Stripe's dashboard, which the engineering team may not notice until C3's card dispute arrives.

The misconception

"The @PollableBean poller tracks which rows have been emitted to the downstream consumer and won't re-poll them until the consumer signals completion. The poller and consumer are part of the same SCS pipeline — the SCS runtime coordinates between them so that in-flight messages are not re-polled."

This is wrong in a fundamental way. The SCS @PollableBean pattern is a fire-and-forget emission mechanism. The Supplier function is invoked by a scheduler thread on the configured fixed-delay and emits its results to the output channel. The scheduler has no visibility into what happens to those messages downstream — it does not track which emitted messages have been acknowledged, which are currently being processed, or which have been published to the target topic. Once the supplier returns and the SCS runtime publishes the messages to the broker (or in-memory channel), the poller's responsibility ends.

The downstream consumer runs independently. It receives messages from the broker and processes them at its own pace. There is no callback mechanism from the consumer back to the poller to say "I have finished processing R1, please exclude it from your next poll." The poller's query condition — WHERE processed_at IS NULL — is the only mechanism for deduplication, and that condition depends on the consumer having called outboxRepo.markProcessed() before the next poll fires.

If the consumer's processing time (Stripe call + DB write) exceeds the poller's fixed-delay, the same row will be re-polled. With a 5-second fixed-delay and a Stripe 3DS call that takes 12 seconds, every 3DS-enrolled customer in a billing run will be re-polled at least once, receiving a second charge.

Diagnosing this failure in production

The hallmark of the @PollableBean re-poll double charge is a pattern where duplicate charges appear for exactly the customers with the longest Stripe response times. If you plot Stripe charge timestamps versus charge amount for a billing run, you will see clusters of duplicates concentrated on high-value international transactions (3DS p99) and sparse duplicates on domestic low-value transactions (fast Stripe responses that complete before the next poll).

Query your billing_outbox table for rows with suspiciously recent processed_at timestamps and cross-check with Stripe:

-- Find outbox rows processed more than once within a billing run window
SELECT outbox_id, subscription_id, created_at, processed_at, stripe_charge_id
FROM billing_outbox
WHERE processed_at BETWEEN '2026-10-24 02:00:00' AND '2026-10-24 03:00:00'
ORDER BY subscription_id, processed_at;

-- Then in Stripe, for any suspicious subscription:
-- stripe charges list --customer <cus_id> --created[gte]=<run_start> --created[lte]=<run_end>
-- Look for multiple ch_ IDs for the same subscription within the billing window.

The fix

The fix requires changes at two levels. At the outbox poll level, mark rows as claimed atomically before emitting them, so the next poll does not re-select them regardless of how long the consumer takes. A common pattern is a claimed_at column set in the same SELECT (using SELECT ... FOR UPDATE SKIP LOCKED in PostgreSQL to handle concurrent poller instances). At the consumer level, use a stable idempotency key derived from the outbox row's stable fields so that if the polling-level deduplication fails (e.g., a crash between claim and commit), the Stripe call is still deduplicated.

@Configuration
public class OutboxPoller {

    private final BillingOutboxRepository outboxRepo;

    @Bean
    @PollableBean
    public Supplier<List<BillingEvent>> outboxPoller() {
        return () -> {
            // Atomic claim: UPDATE billing_outbox SET claimed_at = NOW()
            // WHERE processed_at IS NULL AND claimed_at IS NULL
            // RETURNING *
            // (uses SELECT FOR UPDATE SKIP LOCKED under the hood)
            return outboxRepo.claimUnprocessed(10);
        };
    }
}

@Configuration
public class BillingCharger {

    private final StripeClient stripe;
    private final BillingOutboxRepository outboxRepo;

    @Bean
    public Consumer<BillingEvent> processStripeCharge() {
        return event -> {

            // Stable key from outbox row ID — deterministic across any re-emission
            // outboxId is a UUID assigned at row creation, stable across polls
            String idempotencyKey = sha256(
                event.getOutboxId() + "|" +
                event.getSubscriptionId() + "|" +
                event.getBillingPeriod() + "|" +
                event.getAmountCents()
            );

            // Guard: check for existing charge before calling Stripe
            // (handles crash-and-retry after claim but before mark-processed)
            if (outboxRepo.isAlreadyProcessed(event.getOutboxId())) {
                return; // idempotent skip
            }

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Subscription " + event.getSubscriptionId())
                .build();

            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            outboxRepo.markProcessed(event.getOutboxId(), charge.getId());
        };
    }

    private static String sha256(String input) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }
}

The SQL for the atomic claim with SELECT FOR UPDATE SKIP LOCKED:

-- In BillingOutboxRepository.claimUnprocessed():
WITH claimed AS (
    UPDATE billing_outbox
    SET claimed_at = NOW()
    WHERE id IN (
        SELECT id FROM billing_outbox
        WHERE processed_at IS NULL
          AND claimed_at IS NULL
        ORDER BY created_at ASC
        LIMIT :limit
        FOR UPDATE SKIP LOCKED
    )
    RETURNING *
)
SELECT * FROM claimed;

Additionally, set the poller's fixed-delay to a value that is greater than your worst-case Stripe p99 plus your worst-case DB write p99 plus a safety margin. If Stripe 3DS p99 is 15 seconds and PostgreSQL p99 is 5 seconds, set fixed-delay to at least 30,000ms. This is the defense-in-depth layer — the atomic claim is the primary defense, but the longer poll interval ensures that even if the claim logic has a race condition, re-polling within the Stripe response window is less likely.

# application.yml — fixed poller configuration
spring:
  cloud:
    stream:
      poller:
        # Set to Stripe p99 + DB p99 + safety margin
        fixed-delay: 30000    # 30 s — safely above Stripe 3DS p99 of 15 s
        max-messages-per-poll: 10

Test: outbox row processed exactly once despite second poll before Consumer completes

@SpringBootTest
@Testcontainers
class PollableBeanReemissionTest {

    @Container
    static PostgreSQLContainer<?> postgres =
        new PostgreSQLContainer<>("postgres:16")
            .withDatabaseName("billing_test")
            .withUsername("billing")
            .withPassword("billing");

    @Autowired
    private BillingOutboxRepository outboxRepo;

    @Autowired
    private TestBillingOutboxPublisher publisher; // test helper: directly emits outbox events

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9092));
        wireMock.start();
        // Stripe responds slowly — simulates 3DS latency exceeding poller interval
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withFixedDelay(8_000) // 8s — slower than fixed-delay=5s
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":49900,\"status\":\"succeeded\"}")));
    }

    @Test
    void pollableBean_reemission_stripeCalledWithSameIdempotencyKey() throws Exception {
        String outboxId = UUID.randomUUID().toString();
        BillingEvent event = BillingEvent.builder()
            .outboxId(outboxId)
            .subscriptionId("sub_outbox_test")
            .billingPeriod("2026-10")
            .amountCents(49900L)
            .stripeCustomerId("cus_test")
            .build();

        // Insert row into outbox (simulates transactional outbox write)
        outboxRepo.insertUnprocessed(event);

        // Emit the event twice in rapid succession (simulates two poll firings before markProcessed)
        publisher.emit(event);
        publisher.emit(event); // second emission before consumer finishes

        // Wait for both processing paths to complete (longer than Stripe 8s delay)
        Thread.sleep(20_000);

        // Stripe may be called up to twice but must use the same idempotency key
        List<LoggedRequest> requests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(requests.size()).isGreaterThanOrEqualTo(1);

        Set<String> idempotencyKeys = requests.stream()
            .map(r -> r.getHeader("Idempotency-Key"))
            .collect(Collectors.toSet());
        assertThat(idempotencyKeys).hasSize(1); // stable key — same key on both attempts

        // Outbox row is marked processed exactly once with ch_A
        BillingOutboxRow row = outboxRepo.findById(outboxId);
        assertThat(row.getProcessedAt()).isNotNull();
        assertThat(row.getStripeChargeId()).isEqualTo("ch_A");
    }

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

Eight-item diagnostic checklist for SCS Function model Stripe billing consumers

Run this checklist against any billing consumer that uses the SCS functional programming model before deploying to production.

  1. Function composition definition. If spring.cloud.function.definition contains a | pipe, locate the function that calls Stripe. Confirm its idempotency key is derived from stable payload fields, not UUID.randomUUID(). Remember that retries re-execute the entire composed function — not just the function that threw.
  2. Reactive consumer operator choice. If the billing consumer is a Consumer<Flux<T>>, find every .flatMap() in the pipeline. If it calls Stripe, confirm the idempotency key is computed before the flatMap invocation and is derived from stable event fields. Consider whether concatMap is more appropriate for your ordering requirements.
  3. Poller interval vs. Stripe p99. If the pipeline includes a @PollableBean Supplier, measure your Stripe p99 for the highest-latency card types in your customer base (3DS-enrolled EU and LATAM cards are 10-15s). Confirm spring.cloud.stream.poller.fixed-delay is greater than that p99 plus your DB write p99 plus a safety margin. If not, implement atomic claim-before-emit in the supplier query.
  4. Outbox row claim atomicity. If using the transactional outbox pattern, confirm the supplier uses SELECT ... FOR UPDATE SKIP LOCKED with a claimed_at column update so re-polls cannot select an in-flight row.
  5. Guard-before-Stripe check. Confirm every billing function checks for an existing BillingRecord by the stable idempotency key before calling Stripe. This guard is the second layer of defense when Stripe's own idempotency cache cannot deduplicate (e.g., race between two concurrent calls with the same key within the 30-minute cache window).
  6. DLT message interpretation. If enableDlq=true or dead-letter-policy is configured, document the policy: on-call engineers reading DLT messages must know that the exception header records the failure from the last attempt, not the first — Stripe may have committed on earlier attempts. Never re-ingest a DLT message without first checking Stripe for an existing charge under the stable idempotency key.
  7. Binding name derivation. When using function composition, the SCS auto-derives the binding name as <composed-definition>-in-0 or -out-0. Confirm your application.yml max-attempts, back-off-initial-interval, and DLQ configuration reference the correct derived binding name, not the individual function names — misconfiguration can silently disable retry.
  8. StreamBridge output confirmation. If the pipeline uses StreamBridge.send() to dynamically publish to a downstream topic, confirm that the downstream consumer binding for that topic also uses a stable idempotency key. StreamBridge output does not carry the upstream consumer's retry context — if the upstream consumer retries and re-calls StreamBridge.send(), the downstream topic receives duplicate messages with no binder-level deduplication.

Summary: three SCS Function model Stripe failure modes at a glance

Failure mode Trigger Why UUID_B Fix
Function composition retry DB write throws after ch_A commits; SCS retries the composed binding Retry re-runs the full composed function; processStripeCharge calls UUID.randomUUID() again Stable SHA-256 key in the downstream function; BillingRecord guard before Stripe call
Reactive flatMap concurrent redelivery Kafka rebalance redelivers message into Flux while original is in-flight; flatMap subscribes both concurrently Each concurrent chargeStripe() invocation calls UUID.randomUUID() independently Stable key outside reactive pipeline; concatMap for in-pod ordering; reactive guard before Stripe
@PollableBean re-poll Poller fixed-delay fires again before Consumer marks outbox row as processed Second Consumer invocation for same row calls UUID.randomUUID() independently Atomic claim in poller SQL (SELECT FOR UPDATE SKIP LOCKED); stable key from outbox row ID; increase fixed-delay above Stripe p99

All three failure modes share a common root cause: UUID.randomUUID() as the Stripe idempotency key. A key generated at invocation time is different on every retry, redelivery, and re-poll — which is exactly when Stripe deduplication matters most. A stable key derived from immutable billing event fields is the same on every invocation for the same logical billing event, regardless of how many times the SCS Function runtime invokes the consumer or supplier function. With a stable key and a guard-before-Stripe check, function composition retries, reactive concurrency, and polling re-emissions all become idempotent by construction.

The binder-specific failure modes — Kafka DLT re-ingestion and max.poll.interval.ms eviction, RabbitMQ x-death misread and basicRecover, Pulsar subscriptionType=Shared ackTimeout redelivery and resetCursor replay — are covered in the sibling posts for each binder. The three modes in this post are orthogonal: they apply regardless of which binder you use, because they arise from the application-layer abstractions of the functional programming model, not from the message transport layer. An application using the SCS Kafka Binder with function composition is exposed to both the Kafka-specific failures and the function-model failures described here.

If you are running Keybrake as a proxy between your billing consumer and Stripe, every Stripe call goes through a policy-enforced vault key. Keybrake's audit log captures the idempotency key on every proxied call, making it straightforward to detect duplicate idempotency keys (a symptom of broken retries) versus duplicate stable keys (which should be idempotent). You can also set a daily_usd_cap on the vault key to ensure that a billing loop gone wrong does not charge more than your expected daily billing volume — even if UUID.randomUUID() is used and every retry commits a new charge, the cap will halt charges after a configurable threshold.

Keybrake: a policy layer between your agent and Stripe

Scoped vault keys, per-vendor spend caps, allowed-endpoint allowlists, and a queryable audit log for every Stripe call your agent makes. Detect duplicate charges from SCS Function model retries before the card dispute arrives.