Spring Cloud Stream and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Spring Cloud Stream’s Binder abstraction lets you swap Kafka, RabbitMQ, or Pulsar behind a uniform Consumer<T> programming model. That uniformity hides three framework-level billing failure modes that are invisible at the broker layer and absent from most SCS migration guides — each of which calls stripe.charges.create() a second time for a customer already charged.

This post covers those three failure modes with Java Spring Cloud Stream 3.x functional-model code, content-hash idempotency keys stable across SCS retry attempts and Kafka Binder BATCH redeliveries, per-billing-period vault keys via a spend-cap proxy, and pre-flight database checks — the two-layer governance pattern that closes all three without restructuring your topic topology or switching Binders.

Failure mode 1: Spring Cloud Stream’s default max-attempts=3 retry silently re-invokes the entire Consumer<T> lambda when a downstream exception fires after stripe.charges.create() returns

Spring Cloud Stream enables Spring Retry by default on every consumer binding. The relevant configuration key is spring.cloud.stream.bindings.<binding>.consumer.max-attempts, and its default value is 3. The retry back-off defaults are back-off-initial-interval=1000ms, back-off-multiplier=2.0, and back-off-max-interval=10000ms. Unless a team explicitly sets max-attempts: 1, every Consumer function is silently wrapped in a Spring Retry template that catches any RuntimeException and re-invokes the full lambda up to three times before either publishing to a dead-letter destination or propagating the exception.

The billing failure arises in the gap between a successful Stripe call and a successful downstream write. A billing Consumer<BillingEvent> calls stripe.charges.create() and receives ch_A. It then calls billingRepository.save() to record the charge in PostgreSQL. If save() throws — a database connection pool exhaustion, a transient CannotAcquireLockException, or a DataIntegrityViolationException from a constraint the developer didn’t anticipate — Spring Retry catches the exception and re-invokes the Consumer lambda. The lambda runs again from the top. If it generates a new idempotency key on each invocation — from UUID.randomUUID() or from the SCS-injected KafkaHeaders.DELIVERY_ATTEMPT header — the second invocation calls Stripe with a different key and Stripe creates ch_B.

// application.yml — default SCS consumer retry is active unless explicitly disabled
// spring.cloud.stream.bindings.billing-in-0.consumer.max-attempts=3       (DEFAULT)
// spring.cloud.stream.bindings.billing-in-0.consumer.back-off-initial-interval=1000
// spring.cloud.stream.bindings.billing-in-0.consumer.back-off-multiplier=2.0

@Configuration
public class BillingConfig {

    // UNSAFE: UUID generated inside the Consumer lambda
    // On SCS retry (attempt 2, then 3), a new UUID is generated each time
    // → new idempotency key → stripe.charges.create() → ch_B, ch_C
    @Bean
    public Consumer<BillingEvent> billing(BillingRepository repo, StripeClient stripe) {
        return event -> {
            String requestId = UUID.randomUUID().toString(); // per-invocation — new on retry

            Charge charge = stripe.charges().create(
                ChargeCreateParams.builder()
                    .setAmount(event.getAmountCents())
                    .setCurrency("usd")
                    .setCustomer(event.getCustomerId())
                    .putIdempotencyKey(requestId)   // different key on retry → ch_B
                    .build()
            );

            // If this throws (DB down, connection pool exhausted, unique constraint),
            // Spring Retry catches the exception and re-invokes the lambda.
            // New UUID generated → new Stripe call → ch_B for the same customer.
            repo.save(new BillingRecord(
                event.getCustomerId(),
                event.getBillingPeriod(),
                charge.getId()
            ));
        };
    }
}

The variant using KafkaHeaders.DELIVERY_ATTEMPT is more subtle because it looks intentional. Spring Cloud Stream injects a KafkaHeaders.DELIVERY_ATTEMPT header into each message when the Kafka Binder is active and retry is configured. The header is an integer starting at 1 and incrementing with each retry attempt within the same consumer instance’s Spring Retry loop. A developer who includes this header in the Stripe idempotency key to “distinguish which attempt this Stripe call is from” produces sha256("cust_123:Q3-2026:1") on the first invocation and sha256("cust_123:Q3-2026:2") on the first retry — different keys, Stripe creates ch_B.

// ALSO UNSAFE: KafkaHeaders.DELIVERY_ATTEMPT in the idempotency key
// SCS Kafka Binder injects delivery attempt (1, 2, 3) per Spring Retry loop iteration
@Bean
public Consumer<Message<BillingEvent>> billing(BillingRepository repo) {
    return message -> {
        BillingEvent event = message.getPayload();

        // KafkaHeaders.DELIVERY_ATTEMPT: 1 on first try, 2 on first retry, 3 on second retry
        int attempt = (int) message.getHeaders()
            .getOrDefault(KafkaHeaders.DELIVERY_ATTEMPT, 1);

        // sha256("cust_123:Q3-2026:1") on attempt 1
        // sha256("cust_123:Q3-2026:2") on attempt 2 → ch_B
        // sha256("cust_123:Q3-2026:3") on attempt 3 → ch_C
        String key = DigestUtils.sha256Hex(
            event.getCustomerId() + ":" + event.getBillingPeriod() + ":" + attempt
        ).substring(0, 32);

        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            .putIdempotencyKey(key)
            .build());

        repo.save(new BillingRecord(event.getCustomerId(), event.getBillingPeriod(), charge.getId()));
    };
}

The reason this failure mode is hard to catch in testing is that max-attempts=3 is invisible in the application code — it lives in configuration, not in the Consumer lambda. Developers who write unit tests by calling the Consumer directly (without the SCS retry wrapper) never see the retry behavior. The retry only fires in the SCS runtime when the message is delivered through the actual Binder binding. Integration tests that mock the database call to succeed also hide the issue. The failure surfaces in production when a downstream exception coincides with a completed Stripe call.

The fix for failure mode 1

A content-hash idempotency key derived exclusively from stable business fields is stable across all three SCS retry attempts. The key must produce the same value on attempt 1, attempt 2, and attempt 3 — which means it must not include any value generated inside the Consumer lambda body. Fields that must NOT appear in the key: UUID.randomUUID() (new per lambda invocation), System.currentTimeMillis() or Instant.now() (different millisecond on each retry), KafkaHeaders.DELIVERY_ATTEMPT (increments per retry attempt), Spring Batch’s stepExecution.getId() if billing is embedded in a batch step (changes per run), or any correlation ID generated at the start of the lambda body. The key must be derived solely from the billing intent: customer ID, billing period, and a stable service namespace.

A pre-flight PostgreSQL check using ON CONFLICT DO NOTHING provides a second layer: when Spring Retry re-invokes the lambda on attempt 2, the pre-flight insert finds the row already present from attempt 1’s successful write (or from attempt 1’s write that succeeded before the downstream exception fired) and returns early without calling Stripe.

import org.apache.commons.codec.digest.DigestUtils;
import com.stripe.model.Charge;
import com.stripe.param.ChargeCreateParams;

@Configuration
public class SafeBillingConfig {

    @Bean
    public Consumer<BillingEvent> billing(BillingRepository repo) {
        return event -> {
            String key = makeIdempotencyKey(event.getCustomerId(), event.getBillingPeriod());

            // Pre-flight: claim the slot before calling Stripe.
            // ON CONFLICT DO NOTHING returns 0 rows affected if the key already exists.
            // On SCS retry attempt 2 or 3: key is identical → rowsAffected=0 → return early.
            int rowsAffected = repo.insertIfAbsent(
                event.getCustomerId(), event.getBillingPeriod(), key
            );
            if (rowsAffected == 0) {
                return; // already charged — SCS retry caught by pre-flight
            }

            Charge charge = Charge.create(ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getCustomerId())
                .putIdempotencyKey(key)  // Stripe-side dedup for concurrent-call window
                .build()
            );

            repo.updateChargeId(key, charge.getId());
        };
    }

    static String makeIdempotencyKey(String customerId, String billingPeriod) {
        // Stable fields only — produces the same value on all SCS retry attempts
        // Must NOT include: UUID.randomUUID() (per-invocation),
        // KafkaHeaders.DELIVERY_ATTEMPT (increments per retry),
        // System.currentTimeMillis() (different ms on each attempt),
        // Spring Batch stepExecution.id (changes per batch run)
        String payload = customerId + ":" + billingPeriod + ":scs-billing";
        return DigestUtils.sha256Hex(payload).substring(0, 32);
    }
}

// application.yml — you may also disable SCS retry and handle redelivery at the broker level:
//
// spring:
//   cloud:
//     stream:
//       bindings:
//         billing-in-0:
//           consumer:
//             max-attempts: 1  # disable Spring Retry; exceptions propagate to broker for NACK/requeue

Write-before-call ordering matters here. The pre-flight insert happens before Charge.create(). If the SCS retry fires between the insert and the Stripe call (a thread interrupt, an out-of-memory condition during Stripe parameter serialization), the next attempt finds the pre-flight row and skips Stripe. The customer is not charged — a pending reconciliation job checks status=’pending’ rows older than five minutes and re-issues the Stripe call with the same key.

Failure mode 2: Kafka Binder default BATCH AckMode commits all poll-batch offsets together — a pod crash mid-batch redelivers already-charged customers to the new partition owner

Spring Cloud Stream’s Kafka Binder uses AckMode.BATCH as its default acknowledgment mode. In BATCH mode, the Binder issues a single Kafka offset commit after all records returned by a single poll() call have been processed by the Consumer<T> function. A Kafka poll returns up to max.poll.records records (default: 500, commonly tuned to 10–50 in billing pipelines). If the billing pod processes records 1 through 4 of a 10-record poll batch — calling Charge.create() and writing to the database for each — and then crashes (OOM kill, Kubernetes rolling deploy SIGTERM, or an unhandled exception on record 5), the entire batch is redelivered to the new partition owner on pod restart or partition rebalance. Records 1 through 4 are redelivered despite having already been processed and charged.

Developers who reason from the Consumer<T> perspective expect individual per-invocation ack semantics: the Consumer is called once per record, so an exception on record 5 should only redeliver record 5. This expectation matches AckMode.RECORD behavior but not BATCH. In BATCH mode, what the Kafka Binder sees is a single poll result — and if the batch processing does not complete cleanly (because of an exception that was not caught by Spring Retry, or because the JVM was killed before the batch completion hook ran), all offsets in that batch are left uncommitted.

// application.yml — SCS Kafka Binder default AckMode is BATCH
// NOT configured here = BATCH is active
//
// spring:
//   cloud:
//     stream:
//       kafka:
//         bindings:
//           billing-in-0:
//             consumer:
//               ack-mode: BATCH   # THIS IS THE DEFAULT — often not written in config

@Bean
public Consumer<BillingEvent> billing(BillingRepository repo) {
    return event -> {
        // UNSAFE: no pre-flight check.
        // In BATCH ackMode: billing pod processes records 1-4 of a 10-record poll.
        // Pod crashes between record 4 and the batch offset commit.
        // Kafka redelivers all 10 records to the new partition owner.
        // Records 1-4 already charged → Charge.create() called again → ch_B₁…ch_B₄.
        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            // No idempotency key — redelivery creates a new charge
            .build()
        );
        repo.save(new BillingRecord(
            event.getCustomerId(), event.getBillingPeriod(), charge.getId()
        ));
    };
}

The failure is compounded by a common developer assumption about idempotency key stability. A billing team that knows about Kafka redelivery may add an idempotency key to their Stripe call, but derive it from fields that change between original delivery and redelivery. The most common example: including the Kafka partition or offset in the key under the assumption that “partition + offset uniquely identifies a Kafka record.” Partition and offset are stable across redeliveries of the same record — a record redelivered due to uncommitted BATCH offsets arrives with the same partition and offset — so this particular key would actually be safe. The mistake is when the developer also includes a per-invocation value: a thread-local request ID, a System.currentTimeMillis() captured at handler start, or an attempt counter from KafkaHeaders.DELIVERY_ATTEMPT that was already discussed in failure mode 1. Any per-invocation value in the key defeats the stability guarantee across BATCH redeliveries.

// ALSO UNSAFE: per-invocation timestamp in idempotency key
// Records 1-4 charged with keys containing T=1754300000 (original delivery time)
// On BATCH redelivery: handler starts at T=1754300045 → different key → ch_B

@Bean
public Consumer<BillingEvent> billing(BillingRepository repo) {
    return event -> {
        // handlerStartMs is new on every Consumer invocation, including redeliveries
        long handlerStartMs = System.currentTimeMillis();

        // sha256("cust_123:Q3-2026:1754300000000") on original delivery
        // sha256("cust_123:Q3-2026:1754300045123") on BATCH-redelivery → different key → ch_B
        String key = DigestUtils.sha256Hex(
            event.getCustomerId() + ":" + event.getBillingPeriod() + ":" + handlerStartMs
        ).substring(0, 32);

        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            .putIdempotencyKey(key)
            .build()
        );
        repo.save(new BillingRecord(event.getCustomerId(), event.getBillingPeriod(), charge.getId()));
    };
}

The fix for failure mode 2

There are two independent fixes. The first is to change the Kafka Binder’s ackMode from BATCH to RECORD. In RECORD mode, the Binder commits the offset for each individual record immediately after the Consumer function returns without exception. A pod crash between records 4 and 5 redelivers only records 5 through 10 — records 1 through 4 have their offsets committed and are never redelivered. This is the correct ackMode for billing consumers where each message represents an independent billing intent.

# application.yml — switch Kafka Binder to RECORD ackMode for billing consumers
spring:
  cloud:
    stream:
      kafka:
        bindings:
          billing-in-0:
            consumer:
              ack-mode: RECORD   # commit offset immediately after each Consumer<T> invocation
      bindings:
        billing-in-0:
          group: billing-consumers
          destination: billing-events

The second fix — which should be added regardless of ackMode choice — is the pre-flight database check. Even with RECORD ackMode, a window exists between the Consumer returning successfully and the Binder committing the offset: if the JVM is killed in that sub-millisecond window, the record is redelivered despite having been processed. The pre-flight check closes this window unconditionally.

@Bean
public Consumer<BillingEvent> billing(BillingRepository repo) {
    return event -> {
        String key = makeIdempotencyKey(event.getCustomerId(), event.getBillingPeriod());

        // Pre-flight: closes BATCH ackMode redelivery AND RECORD ackMode edge case.
        // In BATCH mode: records 1-4 committed to DB, batch offset not committed,
        // pod crashes → all 10 redelivered → records 1-4 find existing rows → skip Stripe.
        // In RECORD mode: rare JVM-killed-between-Consumer-return-and-ack → same result.
        int rowsAffected = repo.insertIfAbsent(
            event.getCustomerId(), event.getBillingPeriod(), key
        );
        if (rowsAffected == 0) {
            return; // already charged — BATCH or RECORD redelivery caught by pre-flight
        }

        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            .putIdempotencyKey(key)
            .build()
        );

        repo.updateChargeId(key, charge.getId());
    };
}

// SQL schema
// CREATE TABLE billing_records (
//     customer_id     TEXT NOT NULL,
//     billing_period  TEXT NOT NULL,
//     idempotency_key TEXT NOT NULL,
//     charge_id       TEXT,
//     status          TEXT NOT NULL DEFAULT 'pending',
//     created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
//     CONSTRAINT billing_records_key_uniq UNIQUE (idempotency_key),
//     CONSTRAINT billing_records_period_uniq UNIQUE (customer_id, billing_period)
// );

The UNIQUE (customer_id, billing_period) constraint closes the concurrent-pod race condition: two Consumer instances running on different pods in a consumer group may both fetch the same redelivered record within a rebalance window and both pass an in-memory dedup check. Only one wins the database unique constraint insert. The other sees rowsAffected=0 and returns without calling Stripe.

Failure mode 3: RabbitMQ Binder republish-to-dlq reprocesses billing messages where a Stripe HTTP timeout hid a completed charge

Spring Cloud Stream’s RabbitMQ Binder supports a republish-to-dlq configuration option that, when set to true, catches unhandled Consumer exceptions after all Spring Retry attempts are exhausted and publishes the original message to a dead-letter exchange (by default, <destination>.dlq) with additional AMQP headers: x-exception-message, x-exception-stacktrace, x-original-exchange, and x-original-routing-key. This makes DLQ messages inspectable — operators can read the exception type and stack trace directly from the message headers in the RabbitMQ Management UI.

The DLQ is designed to be a parking lot for genuinely unprocessable messages. In practice, it becomes a routine operational queue: whenever a billing service has an outage and messages accumulate in the DLQ, engineers move messages from billing-events.dlq back to billing-events to reprocess them. This pattern assumes that messages in the DLQ represent pre-Stripe failures — cases where the billing handler never reached the Stripe API call. That assumption is wrong when the exception is a Stripe SDK network timeout.

A Stripe HTTP timeout exception (StripeException with cause java.net.SocketTimeoutException: Read timed out) is thrown by the Stripe Java SDK when the HTTP response from Stripe’s API has not arrived within the SDK’s readTimeout window (default: 80 seconds). The exception fires in the billing Consumer. The Consumer throws. Spring Retry retries up to max-attempts times. Each retry attempt calls Charge.create() with a new key (if the key is per-invocation) and creates a new charge — ch_B, ch_C. After retry exhaustion, the RabbitMQ Binder publishes the message to the DLQ. The DLQ message’s x-exception-stacktrace header shows StripeException: Read timed out. An operator reads this as “Stripe call failed — connection issue” and moves the message back to the main queue for reprocessing. A new Consumer invocation calls Charge.create() — ch_D.

What the operator cannot see in the DLQ message headers is whether Stripe actually created the charge before the timeout. Stripe processes charges server-side. The charge creation logic runs on Stripe’s infrastructure. If Stripe completes the charge object but the HTTP response is lost in transit — due to a network partition between the billing pod’s VPC and Stripe’s API endpoints, a Kubernetes pod network interface reset, or a load balancer that closes the upstream connection after its own timeout fires — ch_A exists in Stripe’s system and the customer has been charged. The SDK throws because it did not receive a response, not because the charge was not created. The DLQ message’s exception header correctly reports the SDK’s experience (timeout), which is accurate but incomplete. Engineers routinely interpret “Stripe SDK threw SocketTimeoutException” as “Stripe did not charge the customer”. It does not mean that.

# application.yml — RabbitMQ Binder DLQ configuration
spring:
  cloud:
    stream:
      rabbit:
        bindings:
          billing-in-0:
            consumer:
              republish-to-dlq: true    # publish failed messages to billing-events.dlq
              auto-bind-dlq: true       # auto-create the DLQ exchange and queue
              dlq-dead-letter-exchange: ""
      bindings:
        billing-in-0:
          group: billing-consumers
          consumer:
            max-attempts: 3

// UNSAFE: no idempotency key — each retry and each DLQ reprocess creates a new charge
@Bean
public Consumer<BillingEvent> billing(BillingRepository repo) {
    return event -> {
        // Attempt 1: Charge.create() sends HTTP request to Stripe.
        // Stripe processes charge, creates ch_A server-side.
        // Network drops before response arrives → StripeException (SocketTimeoutException).
        // Attempt 2 (Spring Retry): Charge.create() → Stripe creates ch_B.
        // Attempt 3 (Spring Retry): Charge.create() → Stripe creates ch_C.
        // Retry exhausted → message published to billing-events.dlq.
        // Operator sees x-exception-stacktrace: "Read timed out" → reprocesses DLQ.
        // DLQ reprocess: Charge.create() → Stripe creates ch_D.
        // Customer charged four times.
        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            // no idempotency key
            .build()
        );
        repo.save(new BillingRecord(event.getCustomerId(), event.getBillingPeriod(), charge.getId()));
    };
}

The DLQ reprocessing variant that uses a content-hash idempotency key but omits the pre-flight check is also vulnerable, specifically beyond Stripe’s 24-hour idempotency window. If a billing event sits in the DLQ for more than 24 hours before an operator processes it — a weekend outage discovered on Monday, a DLQ queue that was not monitored — Stripe’s idempotency cache for the original key has expired. A DLQ reprocess call with the same key creates ch_B. The pre-flight check in the external database has no such expiry: billing_records rows persist indefinitely, and the ON CONFLICT DO NOTHING check returns rowsAffected=0 regardless of whether the row was written 5 minutes or 5 weeks ago.

The fix for failure mode 3

The content-hash idempotency key from failure mode 1 addresses the retry-within-the-Consumer path: all three Spring Retry attempts use the same key, so Stripe returns ch_A on attempts 2 and 3 without creating a new charge. The pre-flight check from failure mode 2 addresses the DLQ reprocessing path: a billing event reprocessed from the DLQ finds the pre-flight row that was inserted during the original successful Stripe call (even if the Consumer threw on the downstream write), returns rowsAffected=0, and exits without calling Stripe. Together they close the full DLQ failure mode.

@Bean
public Consumer<BillingEvent> billing(BillingRepository repo) {
    return event -> {
        String key = makeIdempotencyKey(event.getCustomerId(), event.getBillingPeriod());

        // Pre-flight: covers DLQ reprocessing after Stripe timeout.
        // Original Consumer call: pre-flight insert succeeds (rowsAffected=1),
        // Charge.create() called, ch_A created server-side, HTTP response lost (timeout).
        // Spring Retry attempt 2: same key → Charge.create() → Stripe returns ch_A (cached).
        // Spring Retry attempt 3: same key → ch_A returned again.
        // Retry exhausted → DLQ.
        // DLQ reprocessed (hours or days later):
        //   pre-flight finds existing row → rowsAffected=0 → return without Stripe call.
        //   Stripe's 24h idempotency window irrelevant — DB check has no expiry.
        int rowsAffected = repo.insertIfAbsent(
            event.getCustomerId(), event.getBillingPeriod(), key
        );
        if (rowsAffected == 0) {
            return; // already charged or pre-flight claimed — DLQ reprocess safe
        }

        Charge charge = Charge.create(ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getCustomerId())
            .putIdempotencyKey(key)
            .build()
        );

        // If this write fails, the pre-flight row (status='pending') blocks
        // the next retry from reaching Stripe.
        // A reconciliation job re-issues Charge.create() with the same key
        // for 'pending' rows older than 5 minutes.
        repo.updateChargeId(key, charge.getId());
    };
}

An operational guard is also warranted: annotate DLQ processing runbooks with a reminder that StripeException: Read timed out does not imply the charge was not created. Before moving any billing message from the DLQ back to the main queue, the operator should check the billing_records table for a row with status=’charged’ for the affected customer and billing period. If the row exists, the charge succeeded — the DLQ message should be discarded, not reprocessed.

The complete fix: content-hash idempotency key + vault key + pre-flight check

All three failure modes above share the same root cause: the billing handler calls stripe.charges.create() without a stable idempotency key and without verifying that the charge has not already been created. The content-hash idempotency key and pre-flight check pattern closes all three independently of which failure mode is active. The vault key adds a third layer: even if the first two layers are bypassed — a new Consumer instance that missed the pre-flight row, or a Stripe key cache miss beyond 24 hours — the per-billing-period spend cap prevents the damage from compounding.

import org.apache.commons.codec.digest.DigestUtils;
import com.stripe.model.Charge;
import com.stripe.param.ChargeCreateParams;
import com.stripe.net.RequestOptions;

@Configuration
public class FullySafeBillingConfig {

    @Value("${KEYBRAKE_VAULT_KEY}")
    private String vaultKey;  // per billing period, capped at expected_total * 1.10

    @Bean
    public Consumer<BillingEvent> billing(BillingRepository repo) {
        return event -> {
            String customerId    = event.getCustomerId();
            String billingPeriod = event.getBillingPeriod();
            String key           = makeIdempotencyKey(customerId, billingPeriod);

            // Layer 1: pre-flight claim — if this row already exists, skip.
            // Covers: SCS retry (FM1), Kafka BATCH redelivery (FM2), DLQ reprocessing (FM3).
            int rowsAffected = repo.insertIfAbsent(customerId, billingPeriod, key);
            if (rowsAffected == 0) {
                return; // already charged or in-progress on another consumer instance
            }

            // Layer 2: Stripe idempotency key — broker-side dedup for concurrent-call window.
            // Uses vault key (not raw STRIPE_SECRET_KEY) — vault enforces spend cap, logs each call.
            RequestOptions opts = RequestOptions.builder()
                .setIdempotencyKey(key)
                .setApiKey(vaultKey)  // proxied through spend-cap proxy
                .build();

            Charge charge = Charge.create(ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(customerId)
                .build(),
                opts
            );

            repo.updateChargeId(key, charge.getId());
        };
    }

    static String makeIdempotencyKey(String customerId, String billingPeriod) {
        // Content hash of stable business fields only.
        // Identical across: SCS retry attempts 1/2/3, Kafka BATCH redelivery,
        // RabbitMQ DLQ reprocessing (including >24h after original delivery).
        // Must NOT include: UUID.randomUUID(), System.currentTimeMillis(), Instant.now(),
        // KafkaHeaders.DELIVERY_ATTEMPT, KafkaHeaders.OFFSET (stable but not billing-intent),
        // Spring Batch stepExecution.id, RabbitMQ message ID (may change on DLQ republish).
        String payload = customerId + ":" + billingPeriod + ":scs-billing";
        return DigestUtils.sha256Hex(payload).substring(0, 32);
    }
}

The vault key (KEYBRAKE_VAULT_KEY) is issued per billing period via a spend-cap proxy such as Keybrake. The policy attached to the vault key caps the total spend at expected_total × 1.10. If any combination of the three failure modes above produces duplicate Stripe calls that bypass both the idempotency and pre-flight layers — a new pod starting before the previous one’s pre-flight row was committed, or a Stripe cache miss after 24 hours during DLQ reprocessing — the vault key cap triggers and the proxy returns a 402 before the duplicate charge reaches Stripe. The audit log from the proxy shows the exact vault key, endpoint, customer ID, timestamp, and response for every call, surfacing the duplicate attempt without waiting for a Stripe Dashboard review.

Comparison: which pattern covers which failure mode

Gap analysis: four additional Spring Cloud Stream patterns that warrant attention

1. Functional composition chains: Function<BillingEvent, ChargeResult> output re-processing

Spring Cloud Stream supports reactive and imperative function composition: a Function<BillingEvent, ChargeResult> consumes billing events and produces charge result records to an output binding. If the output binding’s producer call fails (the downstream topic or exchange is unavailable), SCS may retry the entire Function including its stripe.charges.create() input processing. A Function that calls Stripe in its input-processing step and writes to an output channel has the same retry vulnerability as a pure Consumer, but the failure surface is larger: the Function can fail either on the Stripe call or on the output channel write. The content-hash key and pre-flight check must be applied inside the Function’s input processing step, not at the output channel level.

2. Spring Cloud Stream partitioning: partitionKeyExpression and partition reassignment

SCS supports producer-side partitioning via spring.cloud.stream.bindings.<out>.producer.partition-key-expression. When a billing event producer is configured to route by payload.customerId, all events for a given customer land on the same partition. Billing consumers are then guaranteed to see events for the same customer in order. Developers sometimes rely on this ordering guarantee to skip idempotency checks, assuming “only one pod ever processes customer X.” This assumption breaks on consumer group rebalance: during the rebalance window (which can span several seconds), the partition is reassigned. The old pod may be mid-processing a billing event when the partition is revoked; the new pod starts from the last committed offset. If the old pod’s Consumer function completed Charge.create() but did not commit its offset before partition revocation, the new pod redelivers the same event. The content-hash key and pre-flight check handle this redelivery correctly — relying on partition-pinning order instead does not.

3. Polling consumer: PollableMessageSource and manual poll-loop idempotency

Spring Cloud Stream supports a PollableMessageSource pattern where the application controls when to fetch the next message, typically used in slow billing pipelines that rate-limit Stripe API calls. A polling consumer may call source.poll(consumer) in a loop with a configurable interval. If the poll handler calls Charge.create() and then encounters a transient error before calling ack(), the message is redelivered on the next poll cycle. Unlike the functional-model Consumer where SCS retry is framework-managed, the polling consumer’s retry loop is application-managed — developers who implement a simple try/catch retry around poll() without stable idempotency keys will call Stripe multiple times per billing event. The pre-flight check is especially important in polling consumers because the retry interval (and thus the gap between Stripe call and redelivery) is fully configurable and may be very short.

4. CloudEvents headers and ce-id as an idempotency key component

Spring Cloud Stream’s CloudEvents support injects CloudEvents headers into messages when the content type is application/cloudevents+json or when the SCS CloudEvents transformer is active. The ce-id header is a CloudEvents unique event ID — typically a UUID generated at event publication time. Developers who use ce-id as the Stripe idempotency key assume it is stable across redeliveries. For Kafka, this is generally true: ce-id is set in the message headers at publication time and is stable on redelivery. For RabbitMQ, it depends on whether the AMQP publisher preserves the ce-id header on republish — the SCS republishToDlq path preserves all headers, but a manual DLQ-to-main-queue move via rabbitmqadmin may not. Verify header preservation before relying on ce-id for Stripe idempotency. The content-hash key derived from payload business fields is immune to header preservation issues because it requires no header at all.

FAQ

If I set max-attempts: 1, will that prevent duplicate Stripe charges?

Setting max-attempts: 1 disables SCS Spring Retry, which eliminates failure mode 1 (retry re-invokes the Consumer). The exception from a downstream write failure propagates immediately to the Binder, which NACKs or requeues the message at the broker level rather than retrying within the same consumer instance. This shifts the redelivery to the broker layer — which is exactly the right place for at-least-once delivery semantics — but does not eliminate redelivery. The message will be redelivered by Kafka (on BATCH ackMode crash, or on partition rebalance in RECORD mode) or by RabbitMQ (after the NACK). Failure modes 2 and 3 remain active regardless of max-attempts. The content-hash key and pre-flight check are necessary for all three failure modes. max-attempts: 1 is a useful defensive setting for billing consumers, but it is not a substitute for idempotent billing handlers.

Can I use the RabbitMQ message ID (spring_returned_message_correlation) as the Stripe idempotency key?

No. The RabbitMQ message ID (surfaced in SCS as amqp_messageId or the Spring correlation header) may or may not be preserved when a message is republished to the DLQ and later moved back to the main queue. The SCS republish-to-dlq path republishes the original message as a new AMQP message with additional x-exception headers — the new message gets a broker-assigned message ID that differs from the original. If a DLQ-processing tool reads the message, modifies it, and re-publishes to the main exchange, the message ID changes again. A content-hash key from payload business fields is immune to this: it requires no AMQP header and produces the same value from the original message payload and the DLQ-republished message payload because the payload itself is unchanged.

Does Spring Cloud Stream’s Kafka Binder support exactly-once semantics?

Spring Cloud Stream’s Kafka Binder supports Kafka’s producer-side exactly-once semantics via spring.cloud.stream.kafka.bindings.<out>.producer.transactional-id-prefix and idempotent producers. This covers the “produced exactly once to a Kafka topic” guarantee — it does not cover side effects inside the Consumer function, including stripe.charges.create(). Kafka’s transactional API can guarantee that a billing result message is written to an output Kafka topic exactly once, but it cannot guarantee that the external HTTP call to Stripe’s API is made exactly once. The content-hash key and pre-flight check are the correct mechanism for exactly-once Stripe billing semantics regardless of Kafka’s transactional mode.

How should vault keys be scoped in a Spring Cloud Stream billing pipeline?

Issue one vault key per billing period, not per consumer instance, pod, or Spring Retry attempt. A vault key with a policy of {vendor: "stripe", daily_usd_cap: expected_total * 1.10, allowed_endpoints: ["/v1/charges"], expires_at: billing_period_end} caps the total Stripe spend for this billing run at ten percent above the expected amount. If duplicate Stripe calls somehow bypass both the idempotency key and the pre-flight check — two pods racing on the same redelivered message during a Kafka Binder rebalance — the vault cap stops the excess spend at the policy limit. After the cap is hit, the proxy returns a 402 and logs the exact vault key, endpoint, customer ID, and timestamp, giving you the forensic trail to identify the duplicate-call root cause without waiting for a Stripe dispute or customer complaint.

Does switching from @StreamListener to the functional programming model change any of these failure modes?

The functional programming model (Consumer/Function/Supplier beans registered as Spring beans) is the recommended approach in Spring Cloud Stream 3.x. The @StreamListener annotation is deprecated. Both models use the same SCS consumer binding retry configuration (max-attempts, back-off-*) and the same Binder-level delivery semantics. Migrating from @StreamListener to functional beans does not change the retry behavior, the Kafka Binder ackMode, or the RabbitMQ Binder DLQ configuration. All three failure modes described in this post apply to both programming models. If you are migrating from @StreamListener to functional beans and your existing @StreamListener methods were written without idempotency guards, the migration is an opportunity to add the content-hash key and pre-flight check — not a reason to assume the failure modes are resolved by the migration itself.

Related: Apache Kafka and Stripe integration — Kafka Streams and Stripe integration — RabbitMQ and Stripe integration — Apache Pulsar and Stripe integration — Redis Streams and Stripe integration