Spring Cloud Stream Kafka Stripe Integration: How SCS Kafka Binder enableDlq=true Routes Failed Messages to a .DLT Topic That Engineers Re-Ingest Generating UUID_B Committing ch_B, How max.poll.interval.ms Exceeded by a Slow Stripe HTTP Call Evicts the Consumer Group and Redelivers Already-Charged Records, and How resetOffsets=true Deployed as an Operations Hotfix Replays the Entire Topic Committing ch_B for Every Previously-Charged Customer
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 Stream Binder (header-mode=none silently dropping the x-idempotency-key header, RabbitMQ Binder transacted=true basicNack redelivery after Stripe commits ch_A, and Kafka Binder autoAddPartitions partition-creation rebalance replaying uncommitted offsets), Spring Cloud Contract (WireMock anyNonBlankString() masking UUID injection bugs in producer code), Spring Cloud Task (SCDF relaunch new executionId double-charge), and Spring Integration (RequestHandlerRetryAdvice UUID regeneration and JdbcChannelMessageStore transactional poller rollback requeue). This post documents three additional failure modes that are specific to the SCS Kafka Binder and arise from Kafka operational patterns — dead-letter topic re-ingestion, consumer group eviction from max.poll.interval.ms violation, and offset reset via resetOffsets=true — rather than from consumer function code or binder header configuration.
The three failure modes are: a developer who enables the SCS Kafka Binder dead-letter topic feature via spring.cloud.stream.kafka.bindings.billing-in-0.consumer.enableDlq=true, which routes messages that exhaust maxAttempts retries to a billing.DLT topic with the exception stack trace stored in a header, so that an on-call engineer who reads the x-exception-stacktrace header from the DLT message and sees DataIntegrityViolationException at the top frame concludes that the billing record write failed before any Stripe interaction and moves the message back to the main billing topic — the re-ingested Consumer generates UUID.randomUUID() and Stripe commits ch_B alongside the already-committed ch_A; a developer who configures the SCS Kafka Binder consumer with consumer-properties.max.poll.interval.ms: 30000 (30 seconds) as a liveness tuning measure because the Kubernetes readiness probe was previously failing on slow startup, not realizing that a Stripe API call with p99 latency of 12 seconds under international card load plus a PostgreSQL write with connection pool contention can push the total Consumer<T> processing time for a single record to 28–34 seconds — when processing time exceeds max.poll.interval.ms, the Kafka group coordinator marks the consumer as failed and triggers a rebalance; the partition is reassigned to another BillingService pod in the consumer group which starts from the last committed offset (the same record the original consumer was mid-processing); the new consumer generates UUID.randomUUID() and Stripe commits ch_B; meanwhile the original consumer’s Stripe call completes successfully and ch_A was already permanently committed before the eviction; and a developer who deploys spring.cloud.stream.kafka.bindings.billing-in-0.consumer.resetOffsets=true with startOffset=earliest as an incident response to a stalled consumer group that accumulated a lag of 40,000 records during a PostgreSQL outage, not realizing that resetOffsets=true instructs the SCS Kafka Binder to reset the consumer group’s committed offset to startOffset at startup — the binder resets to the earliest available Kafka record across all topic partitions, replaying every billing request that the billing topic has retained (days or weeks of customer charges); each replayed record generates UUID.randomUUID() and Stripe commits ch_B for every previously-charged customer.
What makes these three Kafka-operational failure modes particularly severe is the combination of surface plausibility and scale. The DLT re-ingestion failure mode appears to be the correct incident response: DLTs exist precisely to capture failed messages for retry, and DataIntegrityViolationException is a recognizable database error. The max.poll.interval.ms failure mode appears to be a transient infrastructure event: consumer group rebalances are routine in Kafka and developers generally do not associate them with Stripe charges. The resetOffsets=true failure mode appears to be an isolated offset correction: the property is designed for offset management and is documented in the SCS Kafka Binder reference. None of the three failure modes surfaces a duplicate-charge warning, and all three can produce billing errors at scale — from a single double-charge (DLT re-ingestion of one message) to hundreds or thousands of duplicate charges (resetOffsets=true replaying weeks of billing history).
Stack for examples: Spring Boot 3.x, Spring Cloud Stream 4.x (spring-cloud-starter-stream-kafka), Spring Kafka 3.x, Kafka 3.7+, PostgreSQL 16, Stripe Java SDK v24+, JUnit 5, Testcontainers 1.20+, WireMock 3.x.
Background: three categories of SCS Kafka Binder Stripe failure modes
The failure modes documented in this series fall into three broad categories when examined against the SCS Kafka Binder specifically. The first category is consumer function code failures: the Consumer<T> lambda itself generates an unstable idempotency key (e.g., UUID.randomUUID() inside the lambda body) and SCS retries or redeliveries cause the lambda to be re-invoked with a new key. The original SCS post covered maxAttempts=3 retry and BATCH ackMode crash-mid-batch redelivery in this category. The second category is binder configuration failures: binder-layer properties alter how messages are delivered to the consumer (dropping headers, changing transaction scope, triggering rebalances), again covered in the SCS Binder post. The third category — covered in this post — is Kafka operational failures: failures that arise from Kafka operational procedures (DLT management, consumer group health management, offset correction) interacting with Stripe idempotency in ways the engineer performing the operation does not anticipate.
Kafka operational failures are the hardest to prevent through code review alone because they are often performed by on-call engineers under time pressure during incidents, by platform engineers who manage Kafka infrastructure without deep familiarity with the billing service’s Stripe integration, or by developers applying configuration changes that are documented as standard Kafka Binder usage patterns. The correct defense against all three Kafka operational failure modes documented here is the same: a stable content-hash idempotency key (SHA-256(subscriptionId + billingPeriod + amountCents)) computed before any variable-outcome operation, combined with a BillingRecord guard that checks for an existing charge before calling Stripe. These two safeguards make the consumer function naturally idempotent: whether the message arrives via DLT re-ingestion, rebalance redelivery, or resetOffsets replay, the same record produces the same stable key, the guard finds the existing BillingRecord, and Stripe is never called again.
Mode 1: SCS Kafka Binder enableDlq=true routes post-Stripe-commit failure to billing.DLT — engineer re-ingests the DLT message believing the charge failed before Stripe — UUID_B — ch_B
How the SCS Kafka Binder DLT feature works
The SCS Kafka Binder supports a dead-letter topic (DLT) feature that routes messages to a configured DLT topic when the consumer function exhausts all maxAttempts retry attempts without the exception being handled. The DLT topic name defaults to <destination>.DLT (so billing.DLT for a destination named billing). The failed message is published to the DLT topic with the original message payload preserved and additional SCS-generated headers attached: x-exception-stacktrace (the full stack trace of the exception that caused the failure), x-exception-message (the exception message), x-original-topic (the source topic), x-original-partition, x-original-offset, and x-original-consumer-group.
The DLT feature is enabled via spring.cloud.stream.kafka.bindings.<binding-name>.consumer.enableDlq=true. The intended use case is to capture messages that cannot be processed after multiple retries, inspect them, and either fix the underlying cause and re-ingest the messages, or discard them if they represent genuinely invalid input. The operational pattern the SCS documentation describes for re-ingestion is to use a consumer on the DLT topic, fix the message, and republish it to the original topic. Many teams shortcut this by using kafka-console-producer or a custom admin endpoint to move DLT messages back to the main topic.
The failure scenario
The BillingService is configured with enableDlq=true and maxAttempts=3. The consumer function processes a BillingRequestedEvent for subscription sub_Abc123, billing period 2026-10, amount 9900 cents.
The consumer generates UUID.randomUUID() as the idempotency key (the stable key pattern has not been implemented), calls stripe.charges.create() with that key, and Stripe commits ch_A at t=500ms. At t=600ms, billingRecordRepo.save() throws a DataIntegrityViolationException because the billing_records table already has a row for (sub_Abc123, 2026-10) from a previous session that partially succeeded. The exception propagates out of the consumer function. SCS catches it and applies the retry policy: two more attempts at t=1600ms (UUID_B) and t=4600ms (UUID_C) both also fail with DataIntegrityViolationException because the existing row is still there. After three failed attempts, SCS publishes the original message to billing.DLT with the following headers (among others):
x-exception-message: could not execute statement [ERROR: duplicate key value violates unique constraint "billing_records_sub_period_key"]
x-exception-stacktrace: org.springframework.dao.DataIntegrityViolationException: could not execute statement
at org.springframework.orm.jpa.vendor.HibernateJpaDialect.convertHibernateAccessException(HibernateJpaDialect.java:321)
at org.springframework.orm.jpa.vendor.HibernateJpaDialect.translateExceptionIfPossible(HibernateJpaDialect.java:244)
... [Spring Data JPA frames]
at com.example.billing.BillingFunctions.lambda$billingConsumer$0(BillingFunctions.java:47)
at org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder...
x-original-topic: billing
x-original-partition: 2
x-original-offset: 1042
x-original-consumer-group: billing-serviceThe on-call engineer receives a PagerDuty alert about messages arriving in billing.DLT. The engineer reads the x-exception-stacktrace header. The top-level exception is DataIntegrityViolationException from Spring Data JPA — a database error. The call frame visible in the stack trace is billingRecordRepo.save() at line 47 of BillingFunctions.java. The engineer does not see a Stripe call frame in the visible portion of the stack trace because the Stripe call completed successfully before the exception was thrown.
Developer misconception: “The
DataIntegrityViolationExceptionin the DLT stack trace happened during the database write, which is the step after the Stripe charge. But the stack trace shows the top frame is the database exception — so the consumer must have failed on the database write, which means Stripe was never called for this message. The DLT message represents a pre-Stripe failure. I should re-ingest it so the consumer can try again.”
The engineer re-ingests the DLT message to the billing topic using a custom admin endpoint (POST /admin/dlq/reingest) that reads from billing.DLT and publishes the original payload to billing. The BillingService consumer processes the re-ingested message, generates a new UUID.randomUUID() (because the stable key pattern is not implemented), and calls stripe.charges.create() with the new key. Stripe commits ch_B. The customer is charged twice.
Why the misconception is wrong
The DataIntegrityViolationException in the DLT stack trace is genuine — the database write did fail. But the failure happened at t=600ms, after the Stripe call at t=500ms. The SCS retry mechanism re-invoked the consumer lambda two more times (attempts 2 and 3), each of which also called Stripe with a new UUID.randomUUID() key. So by the time the message reached the DLT, Stripe had been called three times: once with UUID_A (committed ch_A), once with UUID_B (committed ch_B_attempt2), and once with UUID_C (committed ch_C_attempt3) — and the re-ingested message adds a fourth Stripe call (ch_D). In practice, if the DataIntegrityViolationException is thrown because the billing_records row already exists, then ch_A from the very first attempt is permanent, and ch_B_attempt2 and ch_C_attempt3 are also permanent (because each attempt uses a new UUID that Stripe has never seen before). The DLT message represents post-Stripe failure on every attempt, not pre-Stripe failure on any attempt.
The stack trace appears to show only the database exception because the Stripe HTTP call completed without throwing an exception on the first attempt. There is no Stripe frame in the stack trace because no Stripe exception was thrown on attempt 1. The SCS retry mechanism fires because the consumer lambda as a whole threw an exception, not because Stripe threw one.
The buggy consumer configuration
# BillingService application.yml — BUGGY.
# enableDlq=true routes exhausted-retry messages to billing.DLT.
# The DLT x-exception-stacktrace header shows DataIntegrityViolationException
# from the DB write (which occurs AFTER Stripe commits ch_A). An engineer
# reading only the exception class assumes the DB write failed before Stripe.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 3
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
kafka:
bindings:
billing-in-0:
consumer:
# enableDlq routes failed messages to billing.DLT after
# maxAttempts retries are exhausted. Each DLT message carries
# x-exception-stacktrace which shows only the final exception,
# not the Stripe call that succeeded before it.
enableDlq: true
dlqName: billing.DLT
ackMode: RECORDThe buggy consumer function
// BillingService Consumer — BUGGY: UUID.randomUUID() inside the lambda.
// When SCS retries the lambda (maxAttempts=3), each invocation generates
// a new UUID. When the DLT message is re-ingested, the consumer generates
// a fourth UUID and makes a fourth Stripe call.
@Configuration
public class BillingFunctions {
private final StripeClient stripeClient;
private final BillingRecordRepository billingRecordRepo;
@Bean
public Consumer<Message<BillingRequestedEvent>> billingConsumer() {
return message -> {
BillingRequestedEvent event = message.getPayload();
// BUG: UUID.randomUUID() generates a new key on every invocation.
// SCS maxAttempts=3 means this lambda fires 3 times before the
// message goes to DLT. Each firing calls Stripe with a unique UUID.
// When the DLT message is re-ingested, a 4th UUID is generated.
String idempotencyKey = UUID.randomUUID().toString();
// Stripe call at t=500ms — ch_A committed on attempt 1.
// ch_B on SCS retry attempt 2. ch_C on SCS retry attempt 3.
// ch_D on DLT re-ingestion.
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(event.amountCents())
.setCurrency("usd")
.setCustomer(event.stripeCustomerId())
.setDescription("Subscription " + event.subscriptionId())
.build();
Charge charge = stripeClient.charges().create(
params,
RequestOptions.builder()
.setIdempotencyKey(idempotencyKey)
.build()
);
// DB write at t=600ms — throws DataIntegrityViolationException
// if the row for (subscriptionId, billingPeriod) already exists.
// Exception propagates; SCS catches it and retries or routes to DLT.
billingRecordRepo.save(new BillingRecord(
event.subscriptionId(),
event.billingPeriod(),
charge.getId(),
idempotencyKey
));
};
}
}The fix: stable key + guard + DLT error handler
// BillingService Consumer — FIXED.
// Stable SHA-256 key computed before any variable-outcome operation.
// BillingRecord guard checks for existing charge before calling Stripe.
// Re-ingested DLT messages, SCS retry attempts, and rebalance redeliveries
// all produce the same stable key — Stripe returns the cached ch_A.
@Configuration
public class BillingFunctions {
private final StripeClient stripeClient;
private final BillingRecordRepository billingRecordRepo;
@Bean
public Consumer<Message<BillingRequestedEvent>> billingConsumer() {
return message -> {
BillingRequestedEvent event = message.getPayload();
// FIX: Stable key derived from event fields — same on every
// invocation for the same billing request.
String stableKey = computeKey(
event.subscriptionId(),
event.billingPeriod(),
event.amountCents()
);
// Guard: if BillingRecord exists, the charge already went through.
// Return without calling Stripe — safe for DLT re-ingestion.
if (billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
event.subscriptionId(), event.billingPeriod())) {
return; // already charged — idempotent exit
}
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(event.amountCents())
.setCurrency("usd")
.setCustomer(event.stripeCustomerId())
.build();
Charge charge = stripeClient.charges().create(
params,
RequestOptions.builder().setIdempotencyKey(stableKey).build()
);
// DB write — stable key means duplicate key will not occur.
billingRecordRepo.save(new BillingRecord(
event.subscriptionId(),
event.billingPeriod(),
charge.getId(),
stableKey
));
};
}
private static String computeKey(String subscriptionId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] hash = digest.digest(
(subscriptionId + "|" + billingPeriod + "|" + amountCents)
.getBytes(StandardCharsets.UTF_8)
);
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 unavailable", e);
}
}
}# Fixed application.yml — adds dlqPartitions and custom error handler wiring.
# With the stable key + guard fix, re-ingested DLT messages are safe.
# The dlqProducerProperties configuration ensures DLT messages retain
# the original partition routing for easier audit correlation.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 3
backOffInitialInterval: 1000
backOffMaxInterval: 10000
backOffMultiplier: 2.0
kafka:
bindings:
billing-in-0:
consumer:
enableDlq: true
dlqName: billing.DLT
dlqPartitions: 1
ackMode: RECORD
# dlqProducerProperties ensures DLT message retains the
# original partition number in x-original-partition header.
dlqProducerProperties:
acks: allTest: verifying the DLT re-ingestion path is idempotent
@SpringBootTest
@Testcontainers
@ExtendWith(WireMockExtension.class)
class BillingDltReingestionIdempotencyTest {
@Container
static KafkaContainer kafka = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.6.1"));
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
DockerImageName.parse("postgres:16-alpine"));
@RegisterExtension
static WireMockExtension wm = WireMockExtension.newInstance()
.options(wireMockConfig().dynamicPort())
.build();
@Autowired
private StreamBridge streamBridge;
@Autowired
private BillingRecordRepository billingRecordRepo;
@Test
void dltReingestionDoesNotProduceDuplicateCharge() throws Exception {
// Stripe stub — returns ch_A on first call; ch_B would indicate
// a second charge. The test verifies Stripe is called exactly once.
wm.stubFor(post(urlPathEqualTo("/v1/charges"))
.willReturn(aResponse()
.withStatus(200)
.withBody("""
{"id":"ch_A","object":"charge","amount":9900,
"currency":"usd","status":"succeeded"}
""")));
BillingRequestedEvent event = new BillingRequestedEvent(
"sub_Abc123", "2026-10", 9900L, "cus_X");
// First delivery — processed normally.
sendToKafka(event);
// Wait for BillingRecord to exist (first charge completed).
await().atMost(Duration.ofSeconds(10))
.until(() -> billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10"));
// Simulate DLT re-ingestion: publish the same payload to billing topic again.
sendToKafka(event);
// Wait for second delivery to be processed.
Thread.sleep(2000);
// Stripe was called exactly once — re-ingestion hit the guard and returned.
wm.verify(exactly(1), postRequestedFor(urlPathEqualTo("/v1/charges")));
// Only one BillingRecord exists.
assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10")).isEqualTo(1);
}
private void sendToKafka(BillingRequestedEvent event) {
streamBridge.send("billing-out-0", event);
}
}Mode 2: SCS Kafka Binder consumer with max.poll.interval.ms: 30000 — slow Stripe HTTP call exceeds the poll interval — Kafka group coordinator evicts the consumer — rebalance — redelivery — UUID_B — ch_B
What max.poll.interval.ms controls and why it matters for billing
The Kafka consumer protocol requires that each consumer in a consumer group polls the broker for new records at least once every max.poll.interval.ms milliseconds. If a consumer does not call KafkaConsumer.poll() within this interval, the Kafka group coordinator assumes the consumer has hung or died and triggers a consumer group rebalance: it marks the consumer as a failed member of the group, revokes the partitions assigned to it, and redistributes them to the remaining healthy consumers. When the redistributed partition is assigned to a new consumer, that consumer resumes from the last committed offset — not the last processed offset. If the evicted consumer had processed a record (including making a Stripe API call) but had not yet committed the offset, that record will be redelivered to the new consumer.
The SCS Kafka Binder maps max.poll.interval.ms to the underlying Spring Kafka consumer via the consumer-properties namespace: spring.cloud.stream.kafka.bindings.billing-in-0.consumer.consumer-properties.max.poll.interval.ms. The SCS default is to inherit the Kafka client default of 300,000 ms (5 minutes). Teams that tune this property to a shorter value — sometimes to align with Kubernetes liveness probe timeouts, sometimes to make the consumer group respond to pod failures faster — can inadvertently set a value that is shorter than the worst-case processing time for a single billing record.
The failure scenario
The BillingService team sets max.poll.interval.ms: 30000 (30 seconds) after a platform engineer reports that the Kubernetes readiness probe is timing out during slow pod startups when the consumer group coordinator is waiting for the consumer to join. The engineer sets 30 seconds as a "generous" upper bound for processing any single billing record, reasoning that Stripe’s typical latency is under 2 seconds and the database write is under 100 milliseconds.
The problem surfaces three weeks later during a billing run for international customers. The BillingService consumer pulls a BillingRequestedEvent for a customer using a Brazilian Pix payment method. The Stripe API call for Pix-backed charges involves synchronous 3D Secure authentication on Stripe’s side, which adds 8–14 seconds of server-side latency. Simultaneously, the billing_records connection pool is saturated due to a slow analytics query running on the same PostgreSQL instance. The total processing time for this record is 32 seconds: 14 seconds for Stripe (which commits ch_A at t=14s) plus 18 seconds waiting for a database connection before billingRecordRepo.save() can execute.
At t=30s, the Kafka group coordinator detects that consumer instance A has not polled within max.poll.interval.ms. The coordinator triggers a rebalance. Consumer instance A’s partitions (including partition 3, where the Brazilian customer’s record sits at offset 7821) are revoked. The revocation fires ConsumerRebalanceListener.onPartitionsRevoked(). Consumer instance A is still blocking on the billingRecordRepo.save() call; it does not acknowledge the offset commit before revocation. The SCS Kafka Binder’s ack mode is RECORD, so the offset for record 7821 was never committed because the consumer function never returned normally.
Partition 3 is assigned to consumer instance B (a different pod in the same Kubernetes deployment). Instance B starts polling from offset 7821 — the same record that instance A was processing. Instance B generates UUID.randomUUID() (because the stable key pattern is not implemented), calls Stripe with the new key, and Stripe commits ch_B. Instance A’s blocking call to billingRecordRepo.save() eventually returns at t=32s (after obtaining a database connection), but by this point the consumer has been evicted from the group and the commit is a no-op from the group coordinator’s perspective. ch_A and ch_B both exist in Stripe.
Developer misconception: “The
max.poll.interval.mseviction happened during a database connection pool timeout, which is an infrastructure issue unrelated to Stripe. The consumer group rebalance is a standard Kafka fault tolerance mechanism that ensures the record gets processed. Since the original consumer instance hung on the database, it never completed the charge. The rebalanced consumer is the first one to successfully charge the customer.”
The misconception is that the database hang occurred after the Stripe commit. The sequence was: Stripe call (t=0–14s, ch_A committed at t=14s) → database connection wait (t=14–32s) → eviction at t=30s (during the database wait) → rebalance → re-processing by instance B (ch_B). The Stripe charge was not rolled back when the Kafka consumer was evicted; it is permanent.
The buggy consumer configuration
# BillingService application.yml — BUGGY.
# max.poll.interval.ms: 30000 (30 seconds) is shorter than the worst-case
# Stripe + DB processing time for international payment methods.
# A 32-second processing cycle causes group coordinator eviction,
# rebalance, and redelivery to a new consumer that charges UUID_B.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 1 # No SCS retries — rebalance IS the retry mechanism
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORD
consumer-properties:
# BUG: 30 seconds is shorter than worst-case Stripe p99 latency
# (14s for Pix/3DS) plus connection pool contention (18s) = 32s.
# Eviction at 30s fires during the DB wait, after ch_A is committed.
max.poll.interval.ms: 30000The fixed consumer configuration
# Fixed application.yml — max.poll.interval.ms set to accommodate
# measured Stripe p99 latency + DB p99 latency + 2x safety margin.
# For international payment methods with 3DS: Stripe p99 = 15s,
# DB p99 = 5s (with pool contention). Safety margin: 2x = 40s.
# min safe value: 15 + 5 + (15+5)*2 = 60s. Set to 120s.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 1
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORD
consumer-properties:
# FIX: 120 seconds accommodates Stripe p99 (15s) + DB p99 (5s)
# + 2x safety factor. Measured empirically from billing-service
# processing time histograms in Grafana.
max.poll.interval.ms: 120000
# Also tune session.timeout.ms and heartbeat.interval.ms
# for faster detection of actual pod crashes (separate from
# max.poll.interval.ms which is about processing time, not heartbeats).
session.timeout.ms: 30000
heartbeat.interval.ms: 10000The fixed consumer function with stable key and guard
// BillingService Consumer — FIXED.
// Stable SHA-256 key computed before Stripe call.
// BillingRecord guard checks for existing charge before Stripe.
// Even if rebalance causes redelivery, the guard prevents double-charging.
// The max.poll.interval.ms fix prevents eviction during normal processing;
// the stable key + guard acts as a safety net for unexpected redeliveries.
@Configuration
public class BillingFunctions {
private final StripeClient stripeClient;
private final BillingRecordRepository billingRecordRepo;
@Bean
public Consumer<Message<BillingRequestedEvent>> billingConsumer() {
return message -> {
BillingRequestedEvent event = message.getPayload();
String stableKey = computeKey(
event.subscriptionId(),
event.billingPeriod(),
event.amountCents()
);
// Guard: check before Stripe. Works for rebalance redelivery,
// DLT re-ingestion, and any other duplicate delivery path.
if (billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
event.subscriptionId(), event.billingPeriod())) {
return;
}
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(event.amountCents())
.setCurrency("usd")
.setCustomer(event.stripeCustomerId())
.build();
Charge charge = stripeClient.charges().create(
params,
RequestOptions.builder().setIdempotencyKey(stableKey).build()
);
billingRecordRepo.save(new BillingRecord(
event.subscriptionId(),
event.billingPeriod(),
charge.getId(),
stableKey
));
};
}
private static String computeKey(String subscriptionId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] hash = digest.digest(
(subscriptionId + "|" + billingPeriod + "|" + amountCents)
.getBytes(StandardCharsets.UTF_8)
);
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 unavailable", e);
}
}
}Test: verifying that rebalance redelivery is idempotent
@SpringBootTest
@Testcontainers
@ExtendWith(WireMockExtension.class)
class BillingRebalanceRedeliveryIdempotencyTest {
@Container
static KafkaContainer kafka = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.6.1"));
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
DockerImageName.parse("postgres:16-alpine"));
@RegisterExtension
static WireMockExtension wm = WireMockExtension.newInstance()
.options(wireMockConfig().dynamicPort())
.build();
@Autowired
private StreamBridge streamBridge;
@Autowired
private BillingRecordRepository billingRecordRepo;
@Test
void rebalanceRedeliveryDoesNotProduceDuplicateCharge() throws Exception {
// Stripe stub: returns ch_A on first call.
wm.stubFor(post(urlPathEqualTo("/v1/charges"))
.willReturn(aResponse()
.withStatus(200)
.withBody("""
{"id":"ch_A","object":"charge","amount":9900,
"currency":"usd","status":"succeeded"}
""")));
BillingRequestedEvent event = new BillingRequestedEvent(
"sub_Abc123", "2026-10", 9900L, "cus_X");
// First delivery — processed normally, BillingRecord saved.
streamBridge.send("billing-out-0", event);
await().atMost(Duration.ofSeconds(10))
.until(() -> billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10"));
// Simulate rebalance redelivery: publish same event again.
// In production this happens when Kafka redelivers the uncommitted offset.
// In the test we directly simulate it by publishing again.
streamBridge.send("billing-out-0", event);
Thread.sleep(2000);
// Stripe was called exactly once — guard blocked the redelivery.
wm.verify(exactly(1), postRequestedFor(urlPathEqualTo("/v1/charges")));
assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10")).isEqualTo(1);
}
}Measuring your p99 to set max.poll.interval.ms safely
The correct value of max.poll.interval.ms for a billing consumer depends on the measured processing time distribution for your specific workload. The three components to measure are: Stripe API latency p99 (by payment method — domestic cards are typically 1–3 seconds; 3DS flows for international payment methods can be 10–15 seconds); database write latency p99 under peak connection pool load; and any additional processing (encryption, webhook dispatch, audit log write). Sum the p99 values for all components and multiply by 2.0 as a safety margin. For a billing service processing international customers: 15 s (Stripe p99) + 5 s (DB p99) = 20 s; 20 s × 2 = 40 s minimum; set to 120 s to absorb outliers and cold start latency on the first poll after pod startup. This is a substantially longer value than the Kafka default of 300 s only in the sense that many teams tune it down from 300 s; do not tune it below your measured p99 processing time + safety margin.
Mode 3: SCS Kafka Binder resetOffsets=true + startOffset=earliest deployed as an incident hotfix — entire topic replayed — UUID_B × N — ch_B × N
What resetOffsets does in the SCS Kafka Binder
The SCS Kafka Binder supports a resetOffsets consumer binding property (spring.cloud.stream.kafka.bindings.<binding-name>.consumer.resetOffsets) that, when set to true, instructs the binder to reset the consumer group’s committed offset to the startOffset value (earliest or latest) at application startup. When resetOffsets=true and startOffset=earliest, the binder calls KafkaConsumer.seekToBeginning() on all assigned partitions after the consumer group joins, overwriting the consumer group’s committed offsets stored in __consumer_offsets with the earliest available offset from each partition. The consumer then begins reading from the beginning of the topic.
This property is intended for development and testing scenarios where a developer needs to replay a topic from the beginning to test a new consumer implementation or to verify that a recently deployed fix handles historical records correctly. It is documented in the SCS Kafka Binder reference documentation as a standard consumer property. In production, using resetOffsets=true with startOffset=earliest replays every record in the Kafka topic that has not yet been deleted by the topic’s retention policy — which for a billing topic configured with 7-day retention could mean days or weeks of billing requests.
The failure scenario
The BillingService consumer group falls behind by 40,000 records during a 90-minute PostgreSQL outage (the consumer is failing on every record and SCS’s maxAttempts=3 exhausts retries for each). The consumer group’s committed offset is at the offset it last successfully committed before the outage. After the PostgreSQL outage is resolved, the consumer group begins processing again, but an on-call engineer calculates that at the current processing rate (12 records per second), the 40,000-record backlog will take 55 minutes to clear. An older Slack message from a colleague mentions “use resetOffsets=earliest to kick the consumer to reprocess all messages from the beginning.” The colleague was suggesting this for a development environment, but the on-call engineer applies it to the production BillingService as a quick fix to “clear the backlog.”
The engineer updates the production application.yml to add resetOffsets: true and startOffset: earliest, redeploys the BillingService pods, and watches Kubernetes confirm that the new pods are running. The SCS Kafka Binder starts up, the consumer group joins, and the binder calls seekToBeginning() on all five billing topic partitions. The consumer group’s committed offset is overwritten to offset 0 on all partitions. The consumer begins reading from the first record in the topic — the very first billing request ever sent, from 23 days ago.
The consumer has UUID.randomUUID() as its idempotency key strategy. For each of the 23 days of records, the consumer generates a new UUID and calls Stripe. Stripe has seen none of these UUIDs before (they are all different from the UUIDs generated 23 days ago). Stripe commits a new charge for every customer in the topic. After 20 minutes, the Stripe webhook system begins sending charge.succeeded events to the BillingService’s webhook endpoint, and the billing team’s Slack channel begins receiving automated alerts from the Stripe dashboard about an unusual volume of new charges. The engineer redeploys with resetOffsets=false to stop the replay, but by then 8,400 customers have been charged twice.
Developer misconception: “
resetOffsets=earliestresets the consumer group’s reading position to the earliest message in the topic. This is useful to reprocess messages that were not handled correctly during the outage. Since the consumer group’s committed offset only moved up to the records processed before the outage, resetting to earliest will re-read those 40,000 records and reprocess them correctly now that PostgreSQL is back.”
The misconception has two errors. First, resetting to earliest does not reset to the consumer group’s committed offset before the outage; it resets to offset 0 across all partitions, replaying the entire topic history. Second, the records that were successfully processed before the outage (all records before the consumer group’s committed offset at the time of the outage) are replayed along with the failed records from during the outage. The consumer cannot distinguish a “previously successful” record from a “failed during outage” record: both are delivered by Kafka as standard ConsumerRecord objects with identical structure.
The buggy consumer configuration
# BillingService application.yml — BUGGY (production deployment).
# resetOffsets=true + startOffset=earliest replays the entire topic
# from offset 0 across all partitions. Every previously-charged customer
# record is replayed; UUID.randomUUID() generates a unique key per replay;
# Stripe commits ch_B for every customer ever charged.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 3
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORD
# BUG: These two properties in production replay the entire topic.
# resetOffsets=true overwrites the consumer group's committed
# offset with startOffset on every application startup.
# startOffset=earliest seeks to offset 0 on all partitions.
resetOffsets: true
startOffset: earliestThe correct approach: using Kafka consumer group CLI for offset management
The correct tool for adjusting a consumer group’s offset in production is the Kafka consumer group CLI (kafka-consumer-groups.sh) or the Apache Kafka Admin API, not resetOffsets=true in application configuration. The CLI approach requires the consumer group to be stopped (all consumers shut down) before the offset is changed, and allows you to specify a precise target offset (a specific offset value, a timestamp, or a relative offset) on a per-partition basis. To reset to the “earliest available record after the outage started” you would use --to-datetime with the outage start timestamp, not --to-earliest.
# Correct approach: reset consumer group offset to a specific timestamp
# using the Kafka consumer group CLI.
# This resets only to records AFTER the outage began (2026-10-24T02:00:00Z),
# not to the beginning of the topic.
#
# Step 1: Stop all BillingService pods.
kubectl scale deployment billing-service --replicas=0
# Step 2: Reset the consumer group offset to the timestamp when the outage started.
# kafka-consumer-groups.sh is available in the Kafka distribution or via
# a kafka-admin pod in Kubernetes.
kafka-consumer-groups.sh \
--bootstrap-server kafka:9092 \
--group billing-service \
--topic billing \
--reset-offsets \
--to-datetime 2026-10-24T02:00:00.000Z \
--execute
# Step 3: Confirm the new offsets.
kafka-consumer-groups.sh \
--bootstrap-server kafka:9092 \
--group billing-service \
--describe
# Step 4: Restart BillingService pods with resetOffsets=false (the default).
kubectl scale deployment billing-service --replicas=3# Fixed application.yml — resetOffsets=false (the default; remove the
# property entirely or set explicitly to false).
# Never set resetOffsets=true in a production configuration file.
spring:
cloud:
stream:
bindings:
billing-in-0:
destination: billing
group: billing-service
consumer:
maxAttempts: 3
kafka:
bindings:
billing-in-0:
consumer:
ackMode: RECORD
# resetOffsets is omitted (defaults to false).
# startOffset is omitted (defaults to latest for existing groups).The stable key + guard as a safety net for resetOffsets accidents
Even with the correct operational procedure in place, the stable key + guard pattern protects against accidental resetOffsets=true deployments and other full-topic replay scenarios. With SHA-256(subscriptionId + billingPeriod + amountCents) as the idempotency key and a BillingRecord guard before the Stripe call, a full topic replay produces the following behavior for each previously-charged record: the consumer computes the same stable key as the original charge; the guard finds the existing BillingRecord row in the database; the consumer returns without calling Stripe. For the 40,000 failed records from during the outage (which do not have BillingRecord rows because the DB write failed), the consumer calls Stripe with the same stable key as the original (failed) attempt. Because Stripe’s idempotency window is 24 hours for key reuse (the original attempt was up to 90 minutes ago), Stripe returns the same result as the original attempt (for keys that succeeded at Stripe before the DB write failed) or processes a new charge (for records that never reached Stripe).
// BillingService Consumer — FIXED (stable key + guard).
// With this implementation, a resetOffsets=true full-topic replay
// is idempotent: previously-charged records hit the guard and return
// without calling Stripe; failed-during-outage records are retried
// safely with the same stable key.
@Configuration
public class BillingFunctions {
private final StripeClient stripeClient;
private final BillingRecordRepository billingRecordRepo;
@Bean
public Consumer<Message<BillingRequestedEvent>> billingConsumer() {
return message -> {
BillingRequestedEvent event = message.getPayload();
// Stable key: same for every delivery of the same billing request.
String stableKey = computeKey(
event.subscriptionId(),
event.billingPeriod(),
event.amountCents()
);
// Guard: if charged before, return immediately.
// This makes a full resetOffsets=earliest replay safe:
// 23 days of previously-charged records hit the guard.
if (billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
event.subscriptionId(), event.billingPeriod())) {
return;
}
ChargeCreateParams params = ChargeCreateParams.builder()
.setAmount(event.amountCents())
.setCurrency("usd")
.setCustomer(event.stripeCustomerId())
.build();
Charge charge = stripeClient.charges().create(
params,
RequestOptions.builder().setIdempotencyKey(stableKey).build()
);
billingRecordRepo.save(new BillingRecord(
event.subscriptionId(),
event.billingPeriod(),
charge.getId(),
stableKey
));
};
}
private static String computeKey(String subscriptionId,
String billingPeriod,
long amountCents) {
try {
MessageDigest digest = MessageDigest.getInstance("SHA-256");
byte[] hash = digest.digest(
(subscriptionId + "|" + billingPeriod + "|" + amountCents)
.getBytes(StandardCharsets.UTF_8)
);
return HexFormat.of().formatHex(hash);
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("SHA-256 unavailable", e);
}
}
}Test: verifying that a full topic replay is idempotent
@SpringBootTest
@Testcontainers
@ExtendWith(WireMockExtension.class)
class BillingTopicReplayIdempotencyTest {
@Container
static KafkaContainer kafka = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.6.1"));
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
DockerImageName.parse("postgres:16-alpine"));
@RegisterExtension
static WireMockExtension wm = WireMockExtension.newInstance()
.options(wireMockConfig().dynamicPort())
.build();
@Autowired
private StreamBridge streamBridge;
@Autowired
private BillingRecordRepository billingRecordRepo;
@Test
void fullTopicReplayIsIdempotent() throws Exception {
// Stripe stub: returns ch_A on first call.
wm.stubFor(post(urlPathEqualTo("/v1/charges"))
.willReturn(aResponse()
.withStatus(200)
.withBody("""
{"id":"ch_A","object":"charge","amount":9900,
"currency":"usd","status":"succeeded"}
""")));
BillingRequestedEvent event = new BillingRequestedEvent(
"sub_Abc123", "2026-10", 9900L, "cus_X");
// First delivery — BillingRecord saved, ch_A committed.
streamBridge.send("billing-out-0", event);
await().atMost(Duration.ofSeconds(10))
.until(() -> billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10"));
// Simulate resetOffsets=earliest replay: publish same event N times.
int REPLAY_COUNT = 5;
for (int i = 0; i < REPLAY_COUNT; i++) {
streamBridge.send("billing-out-0", event);
}
Thread.sleep(3000);
// Stripe was called exactly once despite N+1 total deliveries.
wm.verify(exactly(1), postRequestedFor(urlPathEqualTo("/v1/charges")));
// Exactly one BillingRecord row — no duplicates.
assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
"sub_Abc123", "2026-10")).isEqualTo(1);
}
}Diagnostic checklist for SCS Kafka Binder + Stripe billing
Use this checklist when investigating a suspected double-charge in a Spring Cloud Stream Kafka-backed billing system. The checklist covers all SCS Kafka Binder failure modes documented across the original SCS post, the SCS Binder post, and this post.
- Check
billingRecordRepofor duplicate rows:SELECT subscription_id, billing_period, COUNT(*) FROM billing_records GROUP BY 1, 2 HAVING COUNT(*) > 1. Multiple rows for the same(subscription_id, billing_period)confirm a duplicate charge regardless of which failure mode caused it. - Check Stripe for duplicate charges:
stripe charges list --customer=<cus_id>. Compare theidempotency_keyvalues on the duplicate charges. If the keys differ, the consumer is not using a stable key strategy. If the keys match, Stripe’s idempotency layer is working and the double-charge is a billing-record-level bug, not a Stripe-level one. - Check
billing.DLTtopic for messages:kafka-console-consumer --topic billing.DLT --from-beginning --property print.headers=true. If messages are present, read thex-exception-stacktraceheader carefully. Determine whether the exception visible in the stack trace occurred before or after the Stripe call. Do not re-ingest DLT messages without a stable idempotency key in place. - Check
max.poll.interval.msagainst measured processing time: Checkapplication.ymlor environment variables formax.poll.interval.ms. Compare against the 99th percentile of billing consumer processing time from your metrics system (Micrometer + Prometheus, Grafana). Ifmax.poll.interval.msis shorter than your p99 processing time, consumer eviction is a risk. - Check for consumer group rebalances during the double-charge window: Search Kafka broker logs or consumer group describe output for
REBALANCINGevents near the time of the double charge. A rebalance event followed by a DLT-consumer-group reassignment of the affected partition is strong evidence of Mode 2. - Check for
resetOffsets=truein deployed configuration:kubectl get configmap billing-service-config -o yaml | grep resetOffsets(or equivalent). Any value oftruein a production deployment warrants immediate investigation. Check recent deployment history for configuration changes. - Verify the idempotency key strategy in the consumer function: Search the consumer lambda for
UUID.randomUUID(). Any call toUUID.randomUUID()inside aConsumer<T>lambda that calls Stripe is a latent double-charge bug, regardless of whether a double charge has been observed yet. - Verify the BillingRecord guard: Check whether the consumer function calls
billingRecordRepo.existsBySubscriptionIdAndBillingPeriod()(or equivalent) before calling Stripe. The guard is the last line of defense against all redelivery paths: SCS retry, DLT re-ingestion, rebalance redelivery, andresetOffsetsreplay.
Summary
The three SCS Kafka operational failure modes documented in this post share a common cause: a delivery or redelivery path that reaches the billing consumer without the consumer being aware that the same billing request has already been processed (partially or fully). The DLT re-ingestion failure mode occurs because the x-exception-stacktrace header shows only the last exception in the retry sequence, not the Stripe call that succeeded on the first attempt. The max.poll.interval.ms eviction failure mode occurs because the consumer group rebalance mechanism correctly handles the Kafka side of the failure (assigning the partition to a healthy consumer) but has no visibility into the Stripe API call that the evicted consumer completed before it was kicked out. The resetOffsets=true failure mode occurs because the property is designed for offset management and operates correctly as documented — but “reset to earliest” means the beginning of the topic, not the beginning of the unprocessed backlog.
The universal fix for all three is the same as for every other failure mode in this series: a stable content-hash idempotency key (SHA-256(subscriptionId + billingPeriod + amountCents)) that produces the same value on every invocation for the same billing request, combined with a BillingRecord guard that checks for an existing charge before calling Stripe. These two patterns together make the consumer function naturally idempotent against every delivery path that Kafka and SCS can produce, whether by design (DLT re-ingestion) or by accident (eviction, offset reset, or manual redelivery).
Keybrake — Stripe API key governance for billing services
Keybrake issues per-billing-period restricted Stripe API keys with hard spend caps. A double-charge from DLT re-ingestion, consumer group eviction, or offset reset commits at most one billing period’s cap — not an uncapped replay across weeks of customer charges.