Spring Cloud Stream Binder Stripe Integration: How header-mode=none Silently Drops the x-idempotency-key Header So the Consumer<T> Falls Back to UUID.randomUUID() and SCS maxAttempts=3 Fires Committing ch_B, How RabbitMQ Binder transacted=true Issues basicNack After Stripe Commits ch_A Causing RabbitMQ Redelivery with UUID_B, and How Kafka Binder autoAddPartitions Triggers a Consumer Group Rebalance That Replays Uncommitted-Offset Records Already Charged
This post is part of the ongoing series on Stripe idempotency failure modes in JVM frameworks. Earlier entries covered Spring Cloud Stream (SCS maxAttempts=3 retry lambda re-invocation, Kafka Binder BATCH ackMode crash-mid-batch redelivery, and RabbitMQ Binder republishToDlq reprocessing with a fresh UUID), Spring Cloud Gateway Filters (AddRequestHeaderGatewayFilter SpEL expression UUID re-evaluation per RetryGatewayFilter attempt, CircuitBreakerGatewayFilter fallback route calling Stripe after the primary route committed ch_A before the response timeout, and WebClient retryWhen re-subscribing the reactive pipeline), Spring Cloud Contract (WireMock consumer stub anyNonBlankString() masking header injection bugs, producer verification happy-path gap, and messaging contract @KafkaListener re-delivery UUID), Spring Cloud Task (SCDF relaunch new executionId, TaskExecutionListener.onTaskFailed() compensation post-commit, and TASK_EXECUTION end_time IS NULL guard misread), and Spring Integration (RequestHandlerRetryAdvice UUID regeneration, JdbcChannelMessageStore transactional poller rollback requeue, and JdbcMessageGroupStore aggregator crash double-release). This post documents three additional Spring Cloud Stream failure modes that operate at the binder configuration layer rather than the consumer function layer — they are silent by design and do not manifest as obvious exceptions in the consumer code itself.
The three failure modes are: a developer who sets spring.cloud.stream.bindings.billing-in-0.consumer.header-mode=none on a SCS Kafka binder consumer binding while attempting to suppress SCS’s contentType-based envelope header injection, not realizing that header-mode=none instructs the Kafka binder to skip all native Kafka record header extraction into the Spring Message<T> headers map — the x-idempotency-key header is physically present in the Kafka record (verifiable via kafka-console-consumer --property print.headers=true) but the SCS consumer never receives it; the consumer reads null for x-idempotency-key and falls back to UUID.randomUUID() on each invocation; when a post-commit DataIntegrityViolationException throws after Stripe commits ch_A, SCS maxAttempts=3 re-invokes the consumer lambda, a fresh UUID_B is generated, and Stripe commits ch_B; a developer who sets spring.cloud.stream.rabbit.bindings.billing-in-0.consumer.transacted=true and annotates the Consumer<Message<Order>> with @Transactional, believing that Spring’s transaction manager will issue a full XA-style rollback of the Stripe charge when the Spring transaction rolls back after a ConstraintViolationException from billingRecordRepo.save(), when in fact Stripe is an external HTTP API with no XA support and the charge is permanent regardless of whether the local transaction commits or rolls back — the Spring AMQP Channel issues basicNack(requeue=true) on transaction rollback, RabbitMQ redelivers the message, and the next consumer invocation generates UUID_B committing ch_B; and a developer who increases spring.cloud.stream.kafka.binder.minPartitionCount from 3 to 6 with autoAddPartitions=true, expecting only a partition count increase on the existing topic, when in fact the new partition creation triggers a Kafka consumer group rebalance for all consumers in the billing service group; consumer instance 1 had processed records at offsets 1000–1004 from partition 2 (committing ch_A₀ through ch_A₄ to Stripe) but had not yet flushed the staged offset commits to Kafka’s __consumer_offsets topic before onPartitionsRevoked fired; partition 2 is reassigned to consumer instance 2 which resumes from the last committed offset 999 and reprocesses records 1000–1004, generating UUID_B₀ through UUID_B₄ and committing ch_B₀ through ch_B₄.
What makes these three binder-layer failures particularly dangerous is that none of them surfaces as an obvious coding error. The first looks like a working consumer with a logging gap — the developer can observe the x-idempotency-key header in Kafka using a console consumer and will assume the BillingService is reading it wrong, not that the SCS Kafka binder is silently discarding all native record headers. The second looks like correct transactional design — the developer is applying a well-understood pattern (annotate a message consumer with @Transactional to get atomic processing) but is applying it to a resource (Stripe) that cannot participate in a JDBC transaction. The third looks like a routine infrastructure scaling operation — increasing partition count is a standard Kafka scaling procedure and most developers are aware that it triggers a rebalance, but fewer realize that the SCS Kafka binder’s default offset commit behavior can leave a window of uncommitted offsets that are replayed under the new partition assignment after the rebalance.
Stack for examples: Spring Boot 3.x, Spring Cloud Stream 4.x (spring-cloud-starter-stream-kafka and spring-cloud-starter-stream-rabbit), Spring Kafka 3.x, Spring AMQP 3.x, Kafka 3.7+, RabbitMQ 3.13+, Stripe Java SDK v24+, JUnit 5, Testcontainers 1.20+, WireMock 3.x.
Background: how SCS Binder configuration properties interact with Stripe idempotency differently than SCS consumer function behavior
Spring Cloud Stream separates the binding layer (the SCS framework infrastructure that connects a Spring Function, Consumer, or Supplier bean to an underlying message broker) from the binder layer (the broker-specific implementation: Kafka Binder, RabbitMQ Binder, Pulsar Binder, etc.). The binding layer is configured through generic properties under spring.cloud.stream.bindings.<binding-name>. The binder layer is configured through broker-specific properties under spring.cloud.stream.kafka.bindings.<binding-name> or spring.cloud.stream.rabbit.bindings.<binding-name>. Both layers expose configuration properties that affect how messages move between the broker and the consumer function, and some of these properties have direct consequences for Stripe idempotency that are invisible at the consumer function level.
The earlier Spring Cloud Stream post documented three failure modes that operate at the consumer function level: maxAttempts=3 re-invokes the Consumer<T> lambda (with a new UUID.randomUUID() call inside the lambda body), the Kafka Binder’s BATCH ackMode redelivers the entire poll batch after a mid-batch pod crash, and the RabbitMQ Binder’s republishToDlq pattern sends already-charged messages to a DLQ from which they are manually re-routed. All three of those failure modes are visible in the consumer function code: the developer can inspect the consumer lambda and see that UUID.randomUUID() is called inside the lambda body, or that there is no BillingRecord guard before the Stripe call. The three failure modes documented in this post are different: the failure is caused by binder configuration properties that alter how messages and headers are delivered to or recovered from the consumer, independently of what the consumer code does with those messages and headers.
The header-mode property controls whether and how SCS extracts native broker-level message headers (Kafka record headers, RabbitMQ AMQP message properties and headers) into the Spring Message<T> headers map that the consumer function receives. The transacted property for the RabbitMQ binder controls whether the Spring AMQP Channel used for message acknowledgment participates in the Spring PlatformTransactionManager’s transaction scope. The autoAddPartitions and minPartitionCount properties for the Kafka binder control whether the SCS framework creates new Kafka partitions at startup and what the minimum partition count should be — properties that can trigger Kafka consumer group rebalances as a side effect of routine configuration changes. Each of these properties operates below the consumer function’s visibility horizon: the consumer does not observe the binder creating a new partition, does not see the Channel issuing a basicNack, and does not receive a signal that header-mode=none has silently dropped its idempotency key header.
The consistent pattern across all three modes is that the binder-layer configuration change introduces a delivery or redelivery path that reaches the consumer with either a missing idempotency key (Mode 1), a requeued-for-redelivery message where the consumer has no knowledge that Stripe already processed the original delivery (Mode 2), or a set of replayed records whose offsets were not committed before the partition was redistributed (Mode 3). In all three cases, the consumer generates a new UUID.randomUUID() (either explicitly as a fallback, or implicitly because it is the consumer’s standard key generation strategy for messages it treats as “new”), and Stripe commits a duplicate charge because the new UUID has never been seen before.
Mode 1: spring.cloud.stream.bindings.billing-in-0.consumer.header-mode=none silently drops the x-idempotency-key message header — Consumer<T> falls back to UUID.randomUUID() per invocation — SCS maxAttempts=3 fires after post-commit DataIntegrityViolationException — UUID_B — ch_B
Setup and the producer side
The billing platform has two services communicating over Kafka. OrderService is the upstream producer: it processes order submissions, computes a deterministic idempotency key, and publishes each order as a Kafka message with the key stored as a native Kafka record header. BillingService is the downstream consumer: it reads the order, uses the idempotency key from the header to call Stripe, and writes a BillingRecord row to its PostgreSQL database.
The OrderService producer code computes the idempotency key as SHA-256(orderId + billingPeriod + amountCents) and stores it as a native Kafka record header named x-idempotency-key. This is a deliberately stable key: the same order with the same billing period and amount will always produce the same key regardless of which OrderService instance generates it or how many times the message is retried at the producer layer.
// OrderService — producer side.
// Computes a deterministic idempotency key and stores it as a native Kafka
// record header via MessageBuilder.setHeader(). SCS Kafka binder will copy
// the Spring Message header into a native Kafka record header if header-mode
// is NOT overridden to 'none' on the consumer binding.
@Service
public class OrderPublisher {
private final StreamBridge streamBridge;
public OrderPublisher(StreamBridge streamBridge) {
this.streamBridge = streamBridge;
}
public void publishOrder(Order order) {
String stableKey = computeIdempotencyKey(
order.orderId(),
order.billingPeriod(),
order.amountCents()
);
Message<Order> message = MessageBuilder
.withPayload(order)
// Store the stable key as a Spring Message header.
// SCS Kafka binder will propagate this into a native Kafka record header
// UNLESS the consumer binding has header-mode=none, which suppresses
// all native header extraction on the consumer side.
.setHeader("x-idempotency-key", stableKey)
.build();
boolean sent = streamBridge.send("billing-out-0", message);
if (!sent) {
throw new MessagePublishException("Failed to send order " + order.orderId());
}
}
static String computeIdempotencyKey(String orderId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
String input = orderId + "|" + billingPeriod + "|" + amountCents;
byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
// Return the full 64-character hex string as the idempotency key.
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 not available", e);
}
}
}The buggy consumer configuration
The BillingService developer encounters an issue during development: the SCS framework is injecting its own contentType header into messages, and the developer is seeing unexpected header values in the Message<Order> headers map that interfere with a custom header-based routing filter. The developer searches the SCS documentation for a property to suppress the contentType header injection and finds header-mode. The documentation says header-mode: none disables header embedding. The developer applies it to the consumer binding, believing it only affects the SCS-internal content-type envelope headers.
# BillingService application.yml — BUGGY configuration.
# header-mode: none was added by the developer to suppress SCS's
# contentType-based envelope header injection.
# CRITICAL BUG: header-mode: none does not selectively suppress only
# contentType headers. It instructs the SCS Kafka binder to bypass ALL
# native Kafka record header extraction. The x-idempotency-key header
# IS present in the Kafka record but will NOT appear in the
# Spring Message<T> headers map received by the Consumer function.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders
group: billing-service
consumer:
# BUG: This disables extraction of ALL native Kafka record headers
# into the Spring Message headers map. The x-idempotency-key header
# that OrderService wrote as a Kafka record header is physically present
# in the Kafka topic (visible via kafka-console-consumer --property
# print.headers=true) but the BillingService Consumer never receives it.
header-mode: none
# SCS default maxAttempts=3 is still active. When the Consumer lambda
# throws DataIntegrityViolationException (post-Stripe-commit DB write
# failure), SCS will retry the lambda 2 more times. Each retry invocation
# has no x-idempotency-key in the headers map and falls back to
# UUID.randomUUID(), generating UUID_B on attempt 2 and UUID_C on attempt 3.
maxAttempts: 3
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORDThe buggy consumer function
// BillingService Consumer — BUGGY because header-mode=none prevents the
// x-idempotency-key header from reaching the Consumer.
// The Consumer correctly reads message.getHeaders().get("x-idempotency-key"),
// but that value will always be null because the SCS Kafka binder dropped
// all native record headers before delivering the message. The null fallback
// generates UUID.randomUUID() per invocation.
@Configuration
public class BillingFunctions {
@Bean
public Consumer<Message<Order>> billingConsumer(BillingService billingService) {
return message -> {
Order order = message.getPayload();
// This read returns null when header-mode=none is configured.
// The x-idempotency-key header is in Kafka but the SCS Kafka binder
// did not extract it. The developer sees null here and assumes the
// OrderService failed to set the header, or that BillingService is
// reading the header name incorrectly.
String idempotencyKey = (String) message.getHeaders()
.get("x-idempotency-key");
if (idempotencyKey == null) {
// BUG: fallback generates a new UUID per Consumer invocation.
// SCS maxAttempts=3 causes this Consumer to be invoked up to 3 times
// for the same Kafka record. Each invocation generates a different UUID.
// Attempt 1: UUID_A — Stripe commits ch_A — billingRecordRepo.save()
// throws DataIntegrityViolationException (duplicate billing record).
// Attempt 2: UUID_B — Stripe commits ch_B — customer charged twice.
// Attempt 3: UUID_C — Stripe commits ch_C — customer charged three times.
idempotencyKey = UUID.randomUUID().toString();
log.warn("x-idempotency-key header missing for order {}; " +
"using fallback UUID {}", order.orderId(), idempotencyKey);
}
billingService.processOrder(order, idempotencyKey);
};
}
}
@Service
public class BillingService {
private final StripeClient stripeClient;
private final BillingRecordRepository billingRecordRepo;
public void processOrder(Order order, String idempotencyKey) {
// Step 1: call Stripe with the idempotency key.
// If idempotencyKey is UUID_A (attempt 1), Stripe commits ch_A.
// If idempotencyKey is UUID_B (attempt 2), Stripe commits ch_B.
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(order.amountCents())
.setCurrency("usd")
.setCustomer(order.stripeCustomerId())
.setDescription("Order " + order.orderId() + " " + order.billingPeriod())
.build();
RequestOptions options = RequestOptions.builder()
.setIdempotencyKey(idempotencyKey)
.build();
Charge charge;
try {
charge = Charge.create(params, options);
} catch (StripeException e) {
throw new BillingException("Stripe charge failed for order " +
order.orderId(), e);
}
// Step 2: write BillingRecord.
// On attempt 1 this succeeds (new record for order.orderId()).
// Wait — actually on attempt 1 this throws DataIntegrityViolationException
// because a previous attempt (from a prior pod restart) already wrote
// a BillingRecord for this orderId. That is the exception that triggers
// SCS maxAttempts retry. On attempt 2, UUID_B → Stripe commits ch_B
// alongside ch_A, and the save still throws — because the unique constraint
// is on orderId, not on idempotencyKey. Third attempt: ch_C.
billingRecordRepo.save(new BillingRecord(
order.orderId(),
order.billingPeriod(),
idempotencyKey,
charge.getId(),
order.amountCents()
));
}
}The failure sequence with timestamps
The orders topic has a message for order ord_001, billing period 2026-10, amount 8900 cents. OrderService computed the stable key KEY_A = "3f7a2b1c..." (64-hex-char SHA-256) and stored it as Kafka record header x-idempotency-key: 3f7a2b1c.... The Kafka record header is verifiable: running kafka-console-consumer --topic orders --property print.headers=true shows x-idempotency-key:3f7a2b1c... alongside the record value. The header is physically present in the Kafka broker storage.
At T=0ms: BillingService consumer instance polls the record. The SCS Kafka binder’s KafkaMessageDrivenChannelAdapter reads the Kafka ConsumerRecord. Because header-mode: none is configured on the consumer binding, the binder’s RecordMessageConverter skips the step that iterates over consumerRecord.headers() and copies each native header into the MessageHeaders map. The binder delivers a Message<Order> to the SCS binding pipeline. That message’s MessageHeaders contains SCS-internal headers (id, timestamp, kafka_offset, kafka_receivedPartitionId, etc.) but does not contain x-idempotency-key.
At T=10ms: the consumer function’s lambda executes. message.getHeaders().get("x-idempotency-key", String.class) returns null. The null check fires. UUID.randomUUID() generates UUID_A = "a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d". The warning log is written.
At T=200ms: BillingService calls Stripe: POST /v1/charges with Idempotency-Key: a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d. Stripe processes the request. At T=700ms: Stripe returns 200 with charge ch_A = "ch_3Qa2B...". The charge has settled.
At T=710ms: billingRecordRepo.save() executes. The billing_records table has a unique constraint on (order_id, billing_period). A previous pod restart had already written a partial BillingRecord for ord_001 / 2026-10 (from a prior processing attempt that succeeded at the DB write but crashed before the Kafka offset was committed). The save() throws DataIntegrityViolationException: ERROR: duplicate key value violates unique constraint "billing_records_order_id_billing_period_key".
At T=720ms: SCS’s Spring Retry interceptor catches the exception. maxAttempts=3 means this is attempt 1 of 3. The interceptor schedules attempt 2 after backOffInitialInterval=1000ms.
At T=1720ms: attempt 2 begins. The SCS binding invokes the consumer lambda again with the same Message<Order> (same Kafka record, same missing x-idempotency-key header because header-mode: none is still in effect). message.getHeaders().get("x-idempotency-key", String.class) returns null again. The null check fires again. UUID.randomUUID() generates UUID_B = "b2c3d4e5-f6a7-8b9c-0d1e-2f3a4b5c6d7e". BillingService calls Stripe with Idempotency-Key: b2c3d4e5-f6a7-8b9c-0d1e-2f3a4b5c6d7e. Stripe has never seen this key and processes a new charge. At T=2220ms: Stripe commits ch_B = "ch_3Rb3C...". The customer has now been charged twice: ch_A ($89.00) and ch_B ($89.00).
The billingRecordRepo.save() on attempt 2 throws again (same constraint violation). SCS schedules attempt 3. At T=6220ms: attempt 3 begins. UUID_C is generated. Stripe commits ch_C. The customer has been charged three times.
Developer misconception: what header-mode=none suppresses
The developer’s mental model was: “I can see the x-idempotency-key header in Kafka using a console consumer with --property print.headers=true. The header exists on the Kafka record. The BillingService must be reading it wrong — maybe the header name has a different case, or the SCS Kafka binder is changing the header name during delivery. The header-mode=none setting I added only suppresses the SCS content-type envelope headers that were polluting my headers map. It should not affect my custom application headers.”
The actual behavior is: header-mode is a binding-level property that controls the entire header embedding and extraction strategy for a binding, not just the SCS internal headers. The three valid values are headers (the default: extract native broker headers into the Spring Message headers map), embeddedHeaders (embed headers in the message payload using SCS’s internal byte-prefix encoding), and none (no header extraction or embedding at all — the consumer receives a Message<T> with only the SCS-internal framework headers and none of the native broker headers). When header-mode: none is configured on a Kafka binder consumer binding, the binder’s DefaultKafkaHeaderMapper is not invoked for incoming records. Every Kafka record header — including application-defined headers like x-idempotency-key — is silently discarded. The header is physically present in Kafka, which explains why kafka-console-consumer shows it. The SCS binder never reads it into the application-layer message.
A secondary misunderstanding is about the relationship between header-mode and spring.cloud.stream.kafka.binder.headerMapperBeanName. A developer who is specifically trying to suppress the SCS contentType header (but not application headers) should use a custom KafkaHeaderMapper bean that filters out only the SCS internal headers, rather than setting header-mode: none which suppresses all header extraction. Alternatively, the developer can simply not rely on the contentType header at the consumer side by configuring the consumer’s contentType property explicitly.
The fix
The immediate fix is to remove header-mode: none from the consumer binding configuration. The default header-mode: headers will extract native Kafka record headers into the Spring Message headers map. The x-idempotency-key header will be available to the consumer function.
# FIXED BillingService application.yml.
# Removed header-mode: none. The default header-mode: headers will extract
# all native Kafka record headers into the Spring Message headers map.
# The x-idempotency-key header from OrderService will be available in the
# consumer's message.getHeaders().get("x-idempotency-key") call.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders
group: billing-service
consumer:
# header-mode: none REMOVED. Default is 'headers' — extracts all
# native Kafka record headers into Spring Message headers map.
maxAttempts: 3
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORDThe belt-and-suspenders fix also adds stable key derivation from the payload as a secondary defense, so that even if headers are misconfigured in the future, the consumer does not fall back to a random UUID but instead derives a deterministic key from the order fields in the message payload. This ensures the key is stable across all SCS retry attempts regardless of header availability:
// FIXED BillingService Consumer.
// Primary key source: x-idempotency-key header (from OrderService, stable SHA-256).
// Secondary key source: derive from payload fields as a fallback.
// Tertiary: never use UUID.randomUUID() — a random UUID is never stable across retries.
@Bean
public Consumer<Message<Order>> billingConsumer(BillingService billingService) {
return message -> {
Order order = message.getPayload();
// Attempt to read the stable key from the header.
String idempotencyKey = (String) message.getHeaders()
.get("x-idempotency-key");
if (idempotencyKey == null || idempotencyKey.isBlank()) {
// Header-mode misconfiguration guard: derive key from payload fields.
// This key is stable across all SCS retry attempts because it is
// computed deterministically from the order's immutable fields.
// It will also match the key that OrderService computed, which means
// Stripe will deduplicate even if the header is somehow lost in transit.
idempotencyKey = OrderPublisher.computeIdempotencyKey(
order.orderId(),
order.billingPeriod(),
order.amountCents()
);
log.warn("x-idempotency-key header missing for order {}; " +
"derived stable key {} from payload fields. " +
"Check header-mode configuration on billing-in-0 binding.",
order.orderId(), idempotencyKey);
}
billingService.processOrder(order, idempotencyKey);
};
}
// ALSO FIXED: billingRecordRepo.save() must use upsert semantics so that
// a DataIntegrityViolationException on an already-written BillingRecord does
// not cause a SCS retry that re-invokes the Stripe call.
// Use INSERT ... ON CONFLICT DO NOTHING or check before insert.
@Service
public class BillingService {
public void processOrder(Order order, String idempotencyKey) {
// Guard: check if BillingRecord already exists for this stable key.
// This prevents double Stripe calls on SCS retry attempts even if
// the stable key fallback was not in place.
Optional<BillingRecord> existing =
billingRecordRepo.findByIdempotencyKey(idempotencyKey);
if (existing.isPresent()) {
// Already charged. Return the existing charge result without calling Stripe.
log.info("BillingRecord already exists for key {}; " +
"charge {} already settled. Skipping Stripe call.",
idempotencyKey, existing.get().stripeChargeId());
return;
}
// Call Stripe with the stable idempotency key.
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(order.amountCents())
.setCurrency("usd")
.setCustomer(order.stripeCustomerId())
.setDescription("Order " + order.orderId() + " " + order.billingPeriod())
.build();
RequestOptions options = RequestOptions.builder()
.setIdempotencyKey(idempotencyKey)
.build();
Charge charge;
try {
charge = Charge.create(params, options);
} catch (StripeException e) {
throw new BillingException("Stripe charge failed", e);
}
// Write BillingRecord. Use save-or-update semantics.
billingRecordRepo.save(new BillingRecord(
order.orderId(),
order.billingPeriod(),
idempotencyKey,
charge.getId(),
order.amountCents()
));
}
}Mode 2: RabbitMQ binder spring.cloud.stream.rabbit.bindings.billing-in-0.consumer.transacted=true — Spring @Transactional Consumer<T> issues AMQP basicNack(requeue=true) when transaction rolls back after Stripe commits ch_A but billingRecordRepo.save() throws ConstraintViolationException — UUID.randomUUID() on RabbitMQ redelivery — UUID_B — ch_B
Setup and the developer’s intent
The team migrates from Kafka to RabbitMQ for the billing service’s message consumption. A developer with strong Spring transactional experience designs the consumer with what appears to be correct atomic processing: the consumer function is annotated with @Transactional, the SCS RabbitMQ binder is configured with transacted=true, and local SCS retry is disabled (maxAttempts=1) in favor of RabbitMQ-level redelivery. The developer’s intent is to achieve all-or-nothing semantics: either Stripe is charged, the BillingRecord is written, and the AMQP message is acknowledged — or the transaction rolls back, the BillingRecord write is undone, and the AMQP message is nacked for redelivery.
# BillingService application.yml — BUGGY RabbitMQ binder configuration.
# transacted=true + @Transactional on the Consumer is intended to give
# XA-style atomicity over the Stripe call + JDBC write + AMQP ack.
# BUG: Stripe is an external HTTP API. It has no XA support and cannot
# participate in the Spring PlatformTransactionManager's transaction.
# When @Transactional rolls back after ConstraintViolationException,
# the JDBC row is removed but the Stripe charge (ch_A) is permanent.
# The AMQP Channel issues basicNack(requeue=true), RabbitMQ redelivers
# the message, and the next Consumer invocation calls Stripe with UUID_B.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders.exchange
group: billing-service
consumer:
# Disable SCS local retry — relying on RabbitMQ-level redelivery instead.
maxAttempts: 1
rabbit:
bindings:
billing-in-0:
consumer:
# BUG: transacted=true makes the Spring AMQP Channel participate in
# the Spring transaction. When the @Transactional consumer throws,
# the transaction rolls back and the Channel issues basicNack(requeue=true).
# This correctly undoes the JDBC write — but it CANNOT undo the Stripe call.
transacted: true
# requeueRejected=true causes nacked messages to be requeued to the
# original queue (not dead-lettered). Combined with transacted=true,
# this means any consumer exception triggers an infinite redelivery loop
# where each delivery generates a new UUID and calls Stripe again.
requeueRejected: trueThe buggy consumer function
// BillingService Consumer — BUGGY with transacted=true + @Transactional.
// The developer expects the Spring transaction to provide atomicity over
// the AMQP ack, the JDBC write, AND the Stripe call. Only the first two
// are JDBC/AMQP resources that Spring's transaction manager can control.
// Stripe cannot participate in a Spring transaction.
@Configuration
public class BillingFunctions {
@Bean
public Consumer<Message<Order>> billingConsumer(BillingService billingService) {
// The @Transactional annotation here applies to the Spring bean method
// invocation. When the billingService.processOrderTransactionally() method
// is called from within a @Transactional context (or the entire
// Consumer lambda is wrapped in a @Transactional proxy by Spring), the
// Spring PlatformTransactionManager begins a transaction before the lambda
// body executes and commits or rolls back after the lambda body completes.
// With transacted=true on the SCS RabbitMQ binder, the AMQP Channel
// participates in the transaction: basicAck is sent on commit, basicNack
// is sent on rollback.
return message -> billingService.processOrderTransactionally(
message.getPayload()
);
}
}
@Service
@Transactional
public class BillingService {
private final BillingRecordRepository billingRecordRepo;
private final StripeGateway stripeGateway;
// BUG: This method is annotated @Transactional (inherited from class-level).
// The Spring transaction:
// - Begins before the method body.
// - Manages the JDBC connection (billingRecordRepo uses JDBC).
// - Manages the AMQP Channel (because transacted=true on the SCS binding).
// - Does NOT manage Stripe (Stripe is HTTP — not an XA resource).
//
// Sequence:
// T=0ms: Spring transaction begins.
// T=100ms: UUID_A = UUID.randomUUID().
// T=500ms: stripeGateway.charge(UUID_A) — Stripe commits ch_A.
// ch_A is now permanent, regardless of local transaction outcome.
// T=600ms: billingRecordRepo.save() — throws ConstraintViolationException.
// T=601ms: @Transactional rolls back.
// JDBC: billing_records row is removed (was written by prior attempt
// that is now being retried — the constraint violation means
// a row already exists for this order).
// AMQP: Channel issues basicNack(requeue=true).
// T=700ms: RabbitMQ redelivers the message.
// T=800ms: Next Consumer invocation: UUID_B = UUID.randomUUID().
// T=1300ms: stripeGateway.charge(UUID_B) — Stripe commits ch_B. DOUBLE CHARGE.
public void processOrderTransactionally(Order order) {
// BUG: UUID.randomUUID() generates a new key per invocation.
// There is no stable key source to fall back to between deliveries.
String idempotencyKey = UUID.randomUUID().toString();
// Stripe call — NOT part of the Spring transaction.
// If this succeeds and the subsequent JDBC write fails, there is NO mechanism
// to roll back the Stripe charge. The developer believes @Transactional
// "wraps everything" but Stripe is outside the transaction boundary.
String chargeId = stripeGateway.chargeCustomer(
order.stripeCustomerId(),
order.amountCents(),
"Order " + order.orderId() + " " + order.billingPeriod(),
idempotencyKey
);
// JDBC write — IS part of the Spring transaction.
// The unique constraint on (order_id, billing_period) will fail if a prior
// attempt already wrote this row. This triggers @Transactional rollback.
// The rollback removes this invocation's JDBC row (if it was written),
// but it also causes the AMQP Channel to basicNack the message for redelivery.
billingRecordRepo.save(new BillingRecord(
order.orderId(),
order.billingPeriod(),
idempotencyKey,
chargeId,
order.amountCents()
));
}
}The AMQP Channel transaction lifecycle explained
To understand why transacted=true on the SCS RabbitMQ binder causes basicNack on Spring transaction rollback, it is necessary to understand how the Spring AMQP RabbitTemplate and SimpleMessageListenerContainer integrate with Spring’s PlatformTransactionManager. When transacted=true is set on the SCS RabbitMQ binder consumer binding, the SCS framework configures the underlying Spring AMQP SimpleMessageListenerContainer with setChannelTransacted(true). This causes the AMQP channel used for message consumption to be enlisted as a participant in the Spring transaction.
The flow is as follows: (1) the AMQP listener container receives a message delivery from RabbitMQ; (2) before invoking the message listener (i.e., the SCS consumer function), the container begins a Spring transaction by calling PlatformTransactionManager.getTransaction(); (3) the transactional AMQP channel is bound to this transaction; (4) the SCS consumer function executes within the transaction; (5) if the consumer function completes without throwing, the container calls PlatformTransactionManager.commit() — this commits the JDBC transaction and causes the AMQP channel to issue basicAck to RabbitMQ, removing the message from the queue; (6) if the consumer function throws an exception, the container calls PlatformTransactionManager.rollback() — this rolls back the JDBC transaction (removing any rows written in this invocation) and causes the AMQP channel to issue basicNack to RabbitMQ. The requeue=true flag on the basicNack (controlled by the requeueRejected property) causes RabbitMQ to put the message back on the front of the queue for redelivery.
The critical point is step (4): the SCS consumer function executes within the transaction, but Stripe is called during the consumer function. The Spring PlatformTransactionManager manages DataSource connections and (with transacted=true) the AMQP channel. It does not manage Stripe. Stripe’s Charge.create() sends an HTTPS request to api.stripe.com. The request is processed by Stripe’s servers and a charge is permanently settled the moment Stripe returns HTTP 200. There is no two-phase commit protocol between Stripe and the Spring transaction. When the Spring transaction rolls back at step (6), the JDBC row is undone and the AMQP channel nacks the message — but Stripe has already settled ch_A. The rollback cannot reach into Stripe’s systems and reverse the charge.
Developer misconception: Stripe as a transactional resource
The developer’s mental model was: “With transacted=true, if my billing transaction fails the message is automatically NACKed and retried by RabbitMQ. That’s the whole point of transactional messaging — the transaction either fully succeeds (Stripe charged, DB written, message acked) or fully rolls back (nothing committed, message NACKed for retry). Stripe is part of my transaction because I call it inside my @Transactional method.”
The actual behavior is that calling a Stripe API inside a @Transactional method does not make Stripe a participant in the Spring transaction. The Spring PlatformTransactionManager manages resources that implement the XA transaction protocol or are otherwise enlisted via a TransactionSynchronization callback. Stripe is a REST API accessed over HTTPS. It has no XA interface, no two-phase commit, and no rollback API that Spring knows about. When @Transactional annotates a method that calls Stripe, the only thing Spring does is manage the JDBC connection lifecycle before and after the method invocation. Stripe’s charge operation is independent of the JDBC transaction and is not reversed when the JDBC transaction rolls back.
The deeper conceptual error is conflating “runs inside @Transactional” with “is part of the transaction.” Only resources that are explicitly managed by the PlatformTransactionManager (or that register a TransactionSynchronization) participate in the transaction’s commit/rollback lifecycle. HTTP calls to external APIs never participate in the transaction, regardless of where they appear in the call stack relative to @Transactional boundaries.
The fix
The fix has two parts: stable key derivation so that RabbitMQ redeliveries call Stripe with the same key (enabling Stripe deduplication), and a BillingRecord guard so that the consumer does not call Stripe at all for messages that have already been processed.
# FIXED application.yml.
# transacted=true is kept — it still provides AMQP ack/nack atomicity with JDBC.
# But the consumer code must derive a stable idempotency key from the message
# payload so that RabbitMQ redeliveries call Stripe with the SAME key.
# Stripe deduplication then ensures ch_A is returned on the second delivery
# instead of a new ch_B being committed.
#
# requeueRejected is changed to false to dead-letter unprocessable messages
# rather than causing an infinite requeue loop. DLQ messages can be inspected
# and replayed manually (with the correct stable key) rather than automatically
# re-creating duplicate charges.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders.exchange
group: billing-service
consumer:
maxAttempts: 1
rabbit:
bindings:
billing-in-0:
consumer:
transacted: true
# FIXED: requeueRejected: false — send unprocessable messages to DLQ
# instead of requeueing for infinite redelivery.
requeueRejected: false
# Configure dead-letter exchange/routing key for the DLQ.
deadLetterExchange: orders.dlx
deadLetterRoutingKey: billing-service.dlq// FIXED BillingService — stable key from payload + BillingRecord guard.
// The stable key is derived from the order’s immutable fields.
// RabbitMQ redeliveries compute the same key.
// Stripe deduplicates: redelivery with KEY_A returns ch_A (already settled),
// not a new ch_B.
@Service
@Transactional
public class BillingService {
private final BillingRecordRepository billingRecordRepo;
private final StripeGateway stripeGateway;
public void processOrderTransactionally(Order order) {
// FIXED: derive a stable idempotency key from the order’s immutable fields.
// SHA-256(orderId | billingPeriod | amountCents) is the same key that
// OrderService computed when publishing the message. RabbitMQ redeliveries
// of the same message will compute the same key.
// Stripe deduplication: when this key is sent to Stripe on the second delivery,
// Stripe returns ch_A from its idempotency cache instead of creating ch_B.
String idempotencyKey = computeStableKey(
order.orderId(),
order.billingPeriod(),
order.amountCents()
);
// Guard: if a BillingRecord already exists for this key, the charge was
// already settled in a previous delivery. Return early without calling Stripe.
// This guard is belt-and-suspenders: Stripe’s own idempotency cache is the
// primary defense, but checking BillingRecord first avoids even the Stripe
// round-trip for already-processed messages.
Optional<BillingRecord> existing =
billingRecordRepo.findByIdempotencyKey(idempotencyKey);
if (existing.isPresent()) {
log.info("BillingRecord already exists for key {} (order {}) — " +
"charge {} already settled on a prior delivery. Skipping.",
idempotencyKey, order.orderId(), existing.get().stripeChargeId());
return;
}
// Call Stripe. On RabbitMQ redelivery, the stable idempotencyKey ensures
// Stripe returns ch_A from cache instead of creating ch_B.
String chargeId = stripeGateway.chargeCustomer(
order.stripeCustomerId(),
order.amountCents(),
"Order " + order.orderId() + " " + order.billingPeriod(),
idempotencyKey
);
// Write BillingRecord within the Spring transaction.
// On commit: AMQP Channel issues basicAck, removing message from queue.
// On rollback (e.g., transient DB failure): AMQP Channel issues basicNack
// to dead-letter the message (requeueRejected=false). Stripe has been called
// but Stripe’s idempotency cache holds ch_A. When the DLQ message is
// manually replayed with the same idempotencyKey, Stripe returns ch_A — no ch_B.
billingRecordRepo.save(new BillingRecord(
order.orderId(),
order.billingPeriod(),
idempotencyKey,
chargeId,
order.amountCents()
));
}
static String computeStableKey(String orderId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
String input = orderId + "|" + billingPeriod + "|" + amountCents;
byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 not available", e);
}
}
}The fixed version retains transacted=true on the SCS RabbitMQ binder because it is still valuable: it ensures that the AMQP ack is atomic with the JDBC commit. If the JDBC write succeeds but the AMQP ack is lost (e.g., the RabbitMQ connection drops between the write and the ack), the transaction will roll back the JDBC write and the AMQP channel will nack the message for dead-lettering. The stable key ensures that when the DLQ message is manually replayed, Stripe deduplicates the charge. What the fix removes is the dangerous combination of transacted=true with requeueRejected=true and a UUID.randomUUID() key generation strategy. The requeue loop is broken by setting requeueRejected=false and moving unprocessable messages to the DLQ, and the duplicate charge risk is eliminated by deriving a stable key from the payload.
Mode 3: spring.cloud.stream.kafka.binder.autoAddPartitions=true with minPartitionCount increase from 3 to 6 — Kafka consumer group rebalance during partition creation replays uncommitted-offset records already charged with ch_A — Consumer<T> generates UUID.randomUUID() per invocation — UUID_B × N for each replayed record
Setup and the scaling operation
The billing service is processing orders from a Kafka orders topic with 3 partitions. As order volume grows during an end-of-month billing sweep, the team decides to increase parallelism by expanding the topic to 6 partitions. The developer who implements the change knows that adding partitions triggers a Kafka consumer group rebalance, but believes that rebalances only cause partition reassignment — not record replay — because Kafka tracks committed offsets in the __consumer_offsets topic and each consumer will resume from the last committed offset for its reassigned partitions.
# BillingService application.yml — initial configuration (3 partitions).
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders
group: billing-service
consumer:
# maxAttempts=1 — no SCS local retry.
maxAttempts: 1
kafka:
binder:
# autoAddPartitions=true allows the SCS Kafka binder to create
# new Kafka topic partitions at startup if the actual partition count
# is less than minPartitionCount. This is convenient for dev/staging
# but has a dangerous side effect in production: creating new partitions
# triggers a Kafka consumer group rebalance for ALL consumers in the
# billing-service group, which can cause uncommitted-offset records to
# be replayed from the last committed offset.
autoAddPartitions: true
minPartitionCount: 3 # Will be changed to 6
bindings:
billing-in-0:
consumer:
# RECORD ackMode: individual offset commits per processed record.
# The commit is sent to Kafka after the Consumer lambda returns.
# BUT: offset commits are batched by the Kafka client and may not
# be flushed to the __consumer_offsets topic immediately after each
# individual record is processed. The Kafka consumer’s internal
# commit buffer accumulates staged commits until either the
# auto.commit.interval.ms fires or an explicit commitSync() is called.
ackMode: RECORD# CHANGED: minPartitionCount increased from 3 to 6.
# At the next BillingService pod startup, the SCS Kafka binder will call
# AdminClient.createPartitions() to increase the orders topic to 6 partitions.
# New partition creation triggers a Kafka consumer group rebalance.
spring:
cloud:
stream:
kafka:
binder:
autoAddPartitions: true
minPartitionCount: 6 # CHANGED from 3 — triggers partition creation
# at next startup, which triggers rebalance.The buggy consumer and offset commit behavior
// BillingService Consumer — BUGGY for the rebalance scenario.
// The consumer generates UUID.randomUUID() per invocation with no stable key
// derivation and no BillingRecord pre-check. When the Kafka rebalance causes
// records at offsets 1000-1004 to be replayed by consumer instance 2 after
// consumer instance 1 had already processed them (committed ch_A0-ch_A4 to Stripe),
// consumer instance 2 generates UUID_B0-UUID_B4 and commits ch_B0-ch_B4.
@Configuration
public class BillingFunctions {
@Bean
public Consumer<Message<Order>> billingConsumer(BillingService billingService) {
return message -> {
Order order = message.getPayload();
// BUG: UUID.randomUUID() per invocation.
// Records 1000-1004 are processed by consumer instance 1 with UUID_A0-A4.
// Kafka rebalance causes records 1000-1004 to be replayed by consumer
// instance 2 (because instance 1’s staged offset commits for 1000-1004
// were not flushed before onPartitionsRevoked fired).
// Consumer instance 2 processes them with UUID_B0-B4 — new keys that
// Stripe has never seen — and Stripe commits ch_B0-ch_B4.
String idempotencyKey = UUID.randomUUID().toString();
billingService.processOrder(order, idempotencyKey);
};
}
}The failure sequence with partition rebalance timeline
The orders topic initially has 3 partitions (P0, P1, P2). The billing service has 3 consumer instances in the billing-service group, each assigned one partition. Consumer instance 1 is assigned to P2. It has been processing records continuously and its most recently committed offset for P2 is 999 (offset 999 was the last record for which the Kafka client issued a commitSync() or commitAsync() that was confirmed by the __consumer_offsets coordinator).
At T=0ms: consumer instance 1 is processing a batch from P2 in its poll loop. It processes records at offsets 1000, 1001, 1002, 1003, 1004 in sequence. For each record, the SCS binding calls the consumer lambda, which calls Stripe.create() with UUID_A₀ through UUID_A₄. Stripe commits ch_A₀ through ch_A₄ for each record. The consumer lambda completes successfully for each record. The SCS Kafka binder, operating in RECORD ackMode, stages an offset commit for each processed record in the Kafka consumer client’s internal commit buffer.
At T=600ms: a new BillingService pod starts up. This pod’s application.yml has the updated minPartitionCount: 6. During the SCS Kafka binder startup sequence, the binder calls KafkaAdmin.createOrModifyTopics(). The admin client detects that the orders topic has 3 partitions and minPartitionCount=6 requires 6. The admin client calls AdminClient.createPartitions() to add 3 new partitions (P3, P4, P5) to the orders topic.
At T=700ms: the Kafka broker confirms the partition creation. The partition count change on the orders topic metadata triggers a Kafka consumer group rebalance for all consumers in the billing-service group. Each consumer’s session with the group coordinator receives a rebalance notification. Consumer instance 1’s Kafka consumer client calls its ConsumerRebalanceListener.onPartitionsRevoked(Collection<TopicPartition> revokedPartitions) callback. The SCS Kafka binder registers an internal ConsumerRebalanceListener that, by default, commits the staged offset commits for the revoked partitions before returning from onPartitionsRevoked.
However, the staged offset commits for records 1000–1004 were not confirmed by the broker. The Kafka consumer client’s commitAsync() calls for these offsets were sent but the acknowledgment from the __consumer_offsets coordinator had not yet arrived when onPartitionsRevoked fired. The SCS binder’s internal onPartitionsRevoked implementation attempts a commitSync() for the staged offsets, but this call times out because the Kafka broker is temporarily busy processing the partition creation request and the __consumer_offsets topic write for the same broker is queued. The commitSync() times out and the staged offset commits for 1000–1004 are lost. The last durably committed offset for P2 in __consumer_offsets remains 999.
At T=900ms: the Kafka consumer group rebalance completes. P2 is assigned to consumer instance 2 (the new pod). Consumer instance 2 calls KafkaConsumer.seek() to resume from the last committed offset for P2, which is 999 (as read from __consumer_offsets). Consumer instance 2 polls records starting from offset 1000. The SCS Kafka binder delivers these records to the consumer function.
At T=1000ms: consumer instance 2 processes the record at offset 1000. UUID.randomUUID() generates UUID_B₀. The consumer calls Stripe with Idempotency-Key: UUID_B₀. Stripe has a record of ch_A₀ committed with UUID_A₀, but UUID_B₀ is a different key. Stripe treats this as a new, independent charge request and commits ch_B₀. The same happens for offsets 1001–1004: UUID_B₁ through UUID_B₄ are generated, and Stripe commits ch_B₁ through ch_B₄. Five customers have each been charged twice.
Developer misconception: Kafka partition rebalance and offset replay
The developer’s mental model was: “I increased the partition count for better parallelism. The rebalance is just Kafka’s partition redistribution — it’s not a consumer crash or restart, so records won’t be replayed. Redelivery only happens if a consumer dies and its last committed offset is earlier than its last-processed record. In a rebalance, the consumer commits its offsets before the partitions are revoked, and the new partition owner picks up from where the previous owner left off. Adding partitions is a routine scaling operation.”
The actual behavior is more nuanced. The statement “the consumer commits its offsets before the partitions are revoked” is only true if the onPartitionsRevoked callback successfully calls commitSync() for all staged commits before returning. This synchronous commit is a network call to the __consumer_offsets broker. If the broker is slow or temporarily unavailable at the moment of the rebalance (a situation that is particularly likely when the rebalance was triggered by a partition creation operation that is itself putting load on the broker), the commitSync() may time out and the staged commits will be lost. The Kafka consumer client’s internal commit buffer holds the staged commits in memory; if the JVM process is not involved in a crash, the commits are still staged in memory, but after the partition is revoked, the staged commits for the revoked partition are abandoned because the consumer no longer owns the partition and cannot commit offsets for it.
The secondary misconception is about the interaction between autoAddPartitions=true and the rebalance trigger. Most developers know that a Kafka rebalance can be triggered by a consumer joining or leaving the group. Fewer know that creating new partitions on a topic also triggers a rebalance for all consumers subscribed to that topic, because each consumer’s metadata for the topic is invalidated and the group coordinator redistributes all partitions (including the new ones) among the consumer group members. The rebalance triggered by autoAddPartitions is indistinguishable from a rebalance triggered by a new consumer joining the group, and carries the same risk of uncommitted-offset replay.
The fix
The fix has three layers. The first layer is to implement ConsumerSeekAware in the consumer listener and flush staged offset commits in onPartitionsRevoked. The second layer is to use a stable idempotency key derived from the record’s content fields so that even when Kafka rebalance causes redelivery, the replayed records call Stripe with the same key and Stripe deduplicates. The third layer is to add a BillingRecord pre-check guard in the consumer so that already-processed orders are short-circuited before reaching Stripe regardless of the redelivery mechanism.
// FIXED BillingService Consumer.
// Implements ConsumerSeekAware to commit staged offsets in onPartitionsRevoked.
// Uses stable key derivation from message payload to survive rebalance replay.
// Adds BillingRecord pre-check guard to short-circuit already-processed orders.
@Configuration
public class BillingFunctions implements ConsumerSeekAware {
private ConsumerSeekCallback seekCallback;
@Override
public void registerSeekCallback(ConsumerSeekCallback callback) {
this.seekCallback = callback;
}
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
// FIXED: flush staged offset commits synchronously before returning from
// onPartitionsRevoked. This ensures that the last committed offsets for
// the revoked partitions reflect the last-processed records.
// The next consumer to receive these partitions will start from the
// correct offset — not from a stale committed offset that predates records
// already processed (and charged via Stripe) by this consumer instance.
if (seekCallback != null) {
seekCallback.commitSync();
}
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments,
ConsumerSeekCallback callback) {
// No-op for assignment — default behavior is to resume from last committed offset.
}
@Bean
public Consumer<Message<Order>> billingConsumer(BillingService billingService) {
return message -> {
Order order = message.getPayload();
// FIXED: stable key derived from order payload fields.
// This key is stable across all redeliveries of the same record,
// whether triggered by a Kafka rebalance, a pod restart, or any
// other Kafka-layer redelivery mechanism.
// Stripe’s idempotency cache: when consumer instance 2 reprocesses
// record at offset 1000 with the same stable key that consumer instance 1
// used, Stripe returns ch_A from cache instead of creating ch_B.
String idempotencyKey = computeStableKey(
order.orderId(),
order.billingPeriod(),
order.amountCents()
);
billingService.processOrder(order, idempotencyKey);
};
}
static String computeStableKey(String orderId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
String input = orderId + "|" + billingPeriod + "|" + amountCents;
byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 not available", e);
}
}
}# FIXED application.yml for Mode 3.
# autoAddPartitions is kept but the consumer also implements ConsumerSeekAware
# to flush committed offsets before partition revoke.
# The stable key derivation in the Consumer provides the primary Stripe
# deduplication guarantee independent of offset commit correctness.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: orders
group: billing-service
consumer:
maxAttempts: 1
kafka:
binder:
autoAddPartitions: true
minPartitionCount: 6
bindings:
billing-in-0:
consumer:
ackMode: RECORD
# FIXED: enable-dlq=true to dead-letter unprocessable messages
# instead of retrying indefinitely on transient consumer errors.
enableDlq: true
dlqName: orders.billing-service.dlq
# FIXED: sync-commits=true forces synchronous offset commits after
# each processed record. This reduces the window between the last
# processed record and the last committed offset, minimizing the
# number of records replayed after a rebalance.
# Note: sync commits increase per-record latency but provide stronger
# rebalance safety for billing use cases.
syncCommits: trueIt is important to note that the syncCommits: true configuration reduces but does not eliminate the replay window. A synchronous offset commit after each record means that the committed offset is at most one record behind the last-processed record (vs. potentially many records behind with asynchronous commits batched over auto.commit.interval.ms). However, if the synchronous commit itself fails (e.g., due to the rebalance being in progress when the commit is attempted), the window can still be one record. The stable key derivation is therefore the more important fix: it ensures that even if Kafka rebalance causes replay of records whose offsets were not committed in time, the replayed records will call Stripe with the same key and Stripe will return the cached ch_A instead of committing a new ch_B.
Diagnostic checklist: eight questions for Spring Cloud Stream Binder + Stripe billing
- Is
header-modeset tononeon any consumer binding that receives Stripe idempotency keys as message headers? Check allspring.cloud.stream.bindings.*.consumer.header-modeproperties across all serviceapplication.ymlfiles. If any binding handling billing messages hasheader-mode: none, all native broker headers — including any application-defined idempotency key headers — are silently discarded before the consumer function receives the message. The consumer will receivenullfor those header values and anyUUID.randomUUID()fallback logic will generate a new key per SCS retry invocation. The safe fix is to removeheader-mode: none(restoring the defaultheadersmode) or to use a customKafkaHeaderMapperthat filters only the specific SCS-internal headers you want to suppress. - Does the consumer derive its Stripe idempotency key from the message payload fields, or does it rely solely on a header value that may be absent or changed between deliveries? A consumer that reads
message.getHeaders().get("x-idempotency-key")and uses the value directly without a payload-derived fallback is exposed to any binder configuration error that causes headers to be dropped (Mode 1), any broker-level message re-routing that strips headers (dead-letter queue re-enqueue, manual message editing), or any producer-side bug that omits the header. The safe design is to derive the key independently from the order’s immutable payload fields usingSHA-256(orderId + billingPeriod + amountCents), verify it against the header value if present (and alert on mismatch), and use the payload-derived key as the authoritative source for the Stripe call. - Is
transacted=trueconfigured on any SCS RabbitMQ binder consumer binding that calls Stripe within a@Transactionalmethod? Inspect allspring.cloud.stream.rabbit.bindings.*.consumer.transactedproperties. Iftransacted=trueis present on a binding whose consumer function calls Stripe, verify that the consumer derives a stable idempotency key from the message payload (notUUID.randomUUID()) so that RabbitMQ redeliveries triggered by transaction rollback will call Stripe with the same key and receivech_Afrom Stripe’s idempotency cache rather than committingch_B. Also verify thatrequeueRejectedisfalseso that unprocessable messages are dead-lettered rather than infinitely requeued. - Is
requeueRejected=trueset on any SCS RabbitMQ binder consumer binding that calls Stripe?requeueRejected=truecombined with a consumer that generatesUUID.randomUUID()per invocation creates an infinite double-charge loop: every exception from any step of the billing process (Stripe call, DB write, downstream API call) causes the message to be requeued, the next delivery generates a new UUID, and Stripe commits a new charge. The safe configuration isrequeueRejected=false(dead-letter on exception) combined with stable key derivation so that manually replayed DLQ messages do not create new charges. - Is
autoAddPartitions=trueset on the SCS Kafka binder in any service that calls Stripe inside aConsumer<T>function? Inspectspring.cloud.stream.kafka.binder.autoAddPartitions. If this property istrue, any increase tominPartitionCountwill create new Kafka partitions at startup, which will trigger a consumer group rebalance. Verify that the consumer derives stable idempotency keys from the message payload and that a BillingRecord pre-check guard is in place so that records replayed due to uncommitted offsets at rebalance time are short-circuited before reaching Stripe. Consider moving partition management to an explicit infrastructure-as-code process (Terraform, Ansible, Confluent Cloud API) rather than relying onautoAddPartitionsin a billing service application. - Does the
Consumer<T>function implementConsumerSeekAware.onPartitionsRevokedand callseekCallback.commitSync()before returning? For Kafka binder consumers processing Stripe billing, the window between the last-processed record and the last-committed offset is directly proportional to the number of records that may be replayed after a rebalance. ImplementingonPartitionsRevokedto flush staged offset commits synchronously before the partition is revoked minimizes this window. Combine withsyncCommits: truein the binder configuration for maximum offset commit freshness. - Is there a BillingRecord pre-check guard in the consumer that queries the database using the stable idempotency key before calling Stripe? The BillingRecord guard is the most robust defense against all three binder-layer failure modes: it is independent of the broker, the binder, and the delivery mechanism. A consumer that checks
billingRecordRepo.findByIdempotencyKey(stableKey)before calling Stripe will short-circuit all redeliveries of already-processed records, regardless of whether the redelivery was caused byheader-mode=nonefallback,transacted=trueAMQP nack, orautoAddPartitionsrebalance replay. The guard adds one database read per message but eliminates all duplicate charge risks at the consumer level. - Does the team have an alert or monitoring rule that fires when the same order ID (or the same stable idempotency key) appears in more than one
BillingRecordrow in the database? Despite all the guards described above, duplicateBillingRecordrows are the first observable symptom of a double-charge event at the consumer layer. A SQL query or database trigger that detectsCOUNT(*) > 1 GROUP BY order_id, billing_periodin thebilling_recordstable, combined with a Stripe webhook listener that alerts on duplicate charge amounts for the same customer within a short time window, provides a real-time safety net that can catch double-charge scenarios before customer-facing disputes are raised.
Consistent defense: stable key derivation at the consumer, not binder header propagation
All three binder-layer failure modes share a structural vulnerability: the consumer relies on a key-derivation mechanism that is external to the consumer’s own processing logic — a header that may be silently dropped by a binder configuration property (header-mode=none), a transaction boundary that cannot encompass Stripe (transacted=true), or an offset commitment assumption that may fail under rebalance conditions (autoAddPartitions). When that external mechanism fails or behaves differently than the developer expected, the consumer falls back to UUID.randomUUID() — a key that is, by definition, unique per invocation and therefore never stable across any form of redelivery, retry, or replay. Any redelivery mechanism in SCS (Spring Retry, AMQP nack+requeue, Kafka offset reset) combined with a UUID.randomUUID() key generation strategy will produce a double charge if the first delivery already committed a Stripe charge before the failure that triggered the redelivery.
The consistent defense is to make the consumer self-sufficient for key derivation: compute SHA-256(orderId + billingPeriod + amountCents) from the message payload in the consumer function itself, without depending on a broker-level header that can be dropped, a transactional resource that cannot wrap Stripe, or an offset commitment strategy that can have a replay window. This key is stable because the order’s immutable fields are always present in the message payload regardless of binder configuration. It is deterministic because the SHA-256 function always produces the same output for the same input. It matches the key that the producer computed and stored in the header (so if the header is present and correctly extracted, both keys will agree), which means Stripe’s idempotency cache will deduplicate any SCS-binder-triggered redelivery that calls Stripe with this key. The BillingRecord pre-check adds a second defense layer that short-circuits the Stripe call entirely for already-processed orders, which is especially valuable when the Stripe idempotency cache TTL (24 hours) has expired and an old DLQ message is being manually replayed.
Testing all three binder failure modes
Testing binder-layer failure modes requires test infrastructure that can simulate binder configuration bugs, AMQP transaction rollbacks, and Kafka partition rebalances. The following examples use Testcontainers for Kafka and RabbitMQ and WireMock for Stripe.
// Mode 1 test: verify that header-mode=none causes x-idempotency-key to be null
// in the Consumer, and that the stable key fallback prevents a double charge.
@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = {"orders"})
class HeaderModeNoneStripeIdempotencyTest {
@Autowired
private StreamBridge streamBridge;
@Autowired
private BillingRecordRepository billingRecordRepo;
@MockBean
private StripeGateway stripeGateway;
@Test
void headerModeNoneConsumerUsesStableKeyFallback() throws InterruptedException {
// Arrange: configure StripeGateway mock to return ch_A on any call
// with the stable SHA-256 key.
String orderId = "ord_001";
String billingPeriod = "2026-10";
long amountCents = 8900L;
String expectedStableKey = BillingFunctions.computeStableKey(
orderId, billingPeriod, amountCents);
when(stripeGateway.chargeCustomer(any(), eq(amountCents), any(),
eq(expectedStableKey)))
.thenReturn("ch_A_test");
// Publish a message WITHOUT an x-idempotency-key header.
// Simulates what header-mode=none does: the header is absent from
// the Message<Order> that the Consumer receives.
// In production, the header IS in Kafka but header-mode=none drops it.
// For testing, we simulate the drop by simply not setting the header.
Order order = new Order(orderId, billingPeriod, amountCents,
"cus_TEST123");
streamBridge.send("billing-out-0", MessageBuilder
.withPayload(order)
// Intentionally omit x-idempotency-key header to simulate header-mode=none.
.build());
// Wait for consumer to process the message.
Thread.sleep(2000);
// Assert: StripeGateway was called with the stable key (not UUID.randomUUID()).
verify(stripeGateway).chargeCustomer(
eq("cus_TEST123"), eq(amountCents), any(), eq(expectedStableKey));
// Assert: only one Stripe call was made (no SCS retry double charge).
verify(stripeGateway, times(1))
.chargeCustomer(any(), anyLong(), any(), any());
// Assert: BillingRecord written with stable key.
Optional<BillingRecord> record =
billingRecordRepo.findByIdempotencyKey(expectedStableKey);
assertThat(record).isPresent();
assertThat(record.get().stripeChargeId()).isEqualTo("ch_A_test");
}
}// Mode 2 test: verify that RabbitMQ redelivery after transaction rollback
// uses the same stable key and Stripe deduplicates (returns ch_A, not ch_B).
@SpringBootTest
@Testcontainers
class RabbitTransactedRedeliveryStripeTest {
@Container
static RabbitMQContainer rabbitMQ = new RabbitMQContainer("rabbitmq:3.13-management");
@Container
static WireMockContainer wireMock = new WireMockContainer("wiremock/wiremock:3.3.1");
@DynamicPropertySource
static void configureProperties(DynamicPropertyRegistry registry) {
registry.add("spring.rabbitmq.host", rabbitMQ::getHost);
registry.add("spring.rabbitmq.port", rabbitMQ::getAmqpPort);
registry.add("stripe.api.base-url", wireMock::getBaseUrl);
}
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private BillingRecordRepository billingRecordRepo;
@BeforeEach
void setupStripeWireMock() {
String orderId = "ord_002";
String billingPeriod = "2026-10";
long amountCents = 14900L;
String stableKey = BillingFunctions.computeStableKey(
orderId, billingPeriod, amountCents);
// Stripe returns the SAME charge ID for both the first delivery
// (key=stableKey, first call) and the redelivery (key=stableKey, second call).
// This simulates Stripe’s idempotency cache: same key → same response.
WireMock.stubFor(WireMock.post("/v1/charges")
.withHeader("Idempotency-Key", WireMock.equalTo(stableKey))
.willReturn(WireMock.aResponse()
.withStatus(200)
.withHeader("Content-Type", "application/json")
.withBody("{\"id\":\"ch_A_test\",\"status\":\"succeeded\"," +
"\"amount\":14900}")));
// Stripe returns an error for any call with a different (non-stable) key.
// This would catch UUID.randomUUID() fallback keys in the test.
WireMock.stubFor(WireMock.post("/v1/charges")
.withHeader("Idempotency-Key", WireMock.notMatching(stableKey))
.willReturn(WireMock.aResponse()
.withStatus(400)
.withHeader("Content-Type", "application/json")
.withBody("{\"error\":{\"type\":\"invalid_request_error\"," +
"\"message\":\"test: non-stable key rejected\"}}")));
}
@Test
void redeliveryAfterTransactionRollbackUsesStableKey() throws InterruptedException {
Order order = new Order("ord_002", "2026-10", 14900L, "cus_RMQ456");
// Publish order. First Consumer invocation will succeed for Stripe (ch_A)
// but the BillingRecord save will fail (simulated by a pre-existing record).
// With transacted=true + requeueRejected=false, the message goes to DLQ.
// We then manually requeue from DLQ to verify stable key behavior.
rabbitTemplate.convertAndSend("orders.exchange", "billing-service", order);
Thread.sleep(3000);
// Assert: the Stripe call was made with the stable key (not a random UUID).
// The WireMock stub would have returned 400 for any non-stable key,
// causing a BillingException that would be visible in the test.
Optional<BillingRecord> record = billingRecordRepo.findByOrderId("ord_002");
assertThat(record).isPresent();
assertThat(record.get().stripeChargeId()).isEqualTo("ch_A_test");
// Assert: exactly one Stripe call was made (BillingRecord guard short-circuited
// the second delivery, which would have been a DLQ replay in this test).
WireMock.verify(1, WireMock.postRequestedFor(
WireMock.urlEqualTo("/v1/charges")));
}
}// Mode 3 test: verify that stable key derivation causes Stripe deduplication
// when a Kafka rebalance causes record replay (offset not committed before revoke).
@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = {"orders"})
class KafkaRebalanceReplayStripeDeduplicationTest {
@Autowired
private KafkaTemplate<String, Order> kafkaTemplate;
@MockBean
private StripeGateway stripeGateway;
@Autowired
private BillingRecordRepository billingRecordRepo;
@Test
void rebalanceReplayUsesStableKeyForStripeDeduplication() throws Exception {
String orderId = "ord_003";
String billingPeriod = "2026-10";
long amountCents = 4900L;
String stableKey = BillingFunctions.computeStableKey(
orderId, billingPeriod, amountCents);
// On the first call with stableKey: return ch_A (simulates original processing).
// On subsequent calls with stableKey: return ch_A again (simulates Stripe
// idempotency cache — same key → same charge returned, no new charge created).
when(stripeGateway.chargeCustomer(any(), eq(amountCents), any(),
eq(stableKey)))
.thenReturn("ch_A_test");
// Simulate the scenario where the Consumer processes the same record twice
// (once as the original delivery, once as the rebalance replay).
// In production, the rebalance replay comes from Kafka. In this test,
// we directly invoke the BillingService’s processOrder method twice
// to verify the BillingRecord guard prevents the second Stripe call.
Order order = new Order(orderId, billingPeriod, amountCents, "cus_KFKA789");
// First invocation (original delivery).
// billingService.processOrder calls stripeGateway then saves BillingRecord.
// Mocked stripeGateway returns ch_A_test.
// BillingRecord is written.
kafkaTemplate.send("orders", orderId, order).get();
Thread.sleep(2000);
// Second invocation (rebalance replay of the same record).
// BillingRecord pre-check guard finds the existing record for stableKey.
// processOrder returns early — stripeGateway is NOT called a second time.
kafkaTemplate.send("orders", orderId, order).get();
Thread.sleep(2000);
// Assert: StripeGateway was called exactly ONCE (not twice).
// The BillingRecord guard short-circuited the second invocation.
verify(stripeGateway, times(1))
.chargeCustomer(any(), anyLong(), any(), eq(stableKey));
// Assert: only one BillingRecord exists for orderId.
List<BillingRecord> records = billingRecordRepo.findAllByOrderId(orderId);
assertThat(records).hasSize(1);
assertThat(records.get(0).stripeChargeId()).isEqualTo("ch_A_test");
}
}The test for Mode 3 uses a simplified simulation of the rebalance scenario by publishing two identical order messages. In a production integration test, a more accurate simulation would use Testcontainers with a real Kafka cluster and programmatically add partitions to the topic mid-test using the Kafka admin client, then observe the consumer group rebalance and verify offset commit behavior. The simplified test still validates the most important invariant: the BillingRecord pre-check guard prevents a second Stripe call for an order that was already processed, regardless of whether the second invocation arrives via SCS retry, RabbitMQ redelivery, Kafka rebalance replay, or any other delivery mechanism. When that guard is combined with stable key derivation, the defense is two-layered: even if the BillingRecord is somehow not written before the redelivery arrives (e.g., the DB write failed on the first delivery), the stable key ensures Stripe deduplicates rather than creating a new charge.
For completeness, the Mode 1 test should also include a variant that explicitly verifies that removing header-mode: none from the consumer binding configuration causes the x-idempotency-key header to be correctly extracted into the Spring Message headers map. This can be tested using @EmbeddedKafka with a producer that sets the header via MessageBuilder.setHeader("x-idempotency-key", stableKey) and a consumer that asserts message.getHeaders().get("x-idempotency-key", String.class) != null. If the test passes, the header extraction is working. If the test fails (null header), the developer knows to check for a header-mode: none configuration in the consumer binding.
Stop Spring Cloud Stream binder-layer Stripe double charges at the proxy
Keybrake is an API key proxy for the SaaS APIs your agents and services call — Stripe, Twilio, Resend. Every proxied request is fingerprinted by billing intent. header-mode=none UUID fallback retries, RabbitMQ transacted=true redelivery with fresh UUIDs, and Kafka rebalance-triggered replays all arrive at Keybrake as different keys for the same fingerprint — deduplicated before they reach Stripe. Early access is open.