Spring Kafka Streams DSL and Stripe Integration: How suppress() Re-emits a Window Record After State-Store Crash-Restore and a Billing Consumer Charges with UUID_B, EXACTLY_ONCE_V2 Protects Kafka Writes but Not the .foreach() Stripe Call Inside the Transaction Boundary, and a STREAM_TIME Punctuator Fires Three Times on a Three-Day Timestamp Gap Generating Three Charges

Kafka Streams DSL introduces three deployment and execution patterns that each create a distinct Stripe idempotency failure mode: a windowed aggregation operator that can re-emit the same window record after a state-store crash-restore cycle, an exactly-once processing guarantee that applies to Kafka writes but leaves external HTTP calls exposed, and a stream-time punctuator whose fire frequency is driven by event timestamps rather than wall-clock time. Each failure produces a duplicate Stripe charge for a customer who has already been charged.

This post covers three failure modes that are mechanically distinct from the Kafka Streams exactly-once semantics covered in Spring Kafka Streams EOS and Stripe Integration. That post addressed how disabling or misconfiguring EXACTLY_ONCE_V2 causes duplicate output-topic writes that a billing consumer processes twice. The three modes here arise from different DSL constructs: the suppress() operator’s interaction with state-store checkpointing, the transaction scope boundary of EXACTLY_ONCE_V2, and the punctuation scheduling semantics of PunctuationType.STREAM_TIME. All three can occur even when the developer believes the topology is correctly configured for exactly-once delivery. Prior related posts in this series include Spring Kafka EOS, Micronaut Kafka EOS, Spring Boot Actuator scheduler patterns, and Spring Data R2DBC reactive patterns.

Mode 1: suppress(untilWindowCloses) crash-restore re-emission — state-store replay pushes the window watermark past the end again, suppress() fires a second time, and the billing consumer processes the duplicate record with UUID_B

The setup

The developer builds a daily usage-based billing system using Kafka Streams DSL. Each usage event arrives on the usage-events topic with a key of customerId and a value representing the unit count for that event. The topology aggregates usage per customer per day and emits exactly one billing trigger per customer per day using suppress(Suppressed.untilWindowCloses()):

@Configuration
public class BillingTopologyConfig {

    @Bean
    public KStream<String, Long> buildBillingTopology(StreamsBuilder builder) {
        // Aggregate daily usage per customer
        KTable<Windowed<String>, Long> dailyUsage = builder
            .stream("usage-events", Consumed.with(Serdes.String(), Serdes.Long()))
            .groupByKey()
            .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofDays(1)))
            .count(Materialized.as("daily-usage-store"));

        // suppress() emits exactly once when the window closes
        KStream<Windowed<String>, Long> windowedCounts = dailyUsage
            .suppress(Suppressed.untilWindowCloses(
                BufferConfig.unbounded().shutDownWhenFull()))
            .toStream();

        // Convert to billing trigger format and send to output topic
        windowedCounts
            .map((windowedKey, count) -> KeyValue.pair(
                windowedKey.key(),
                new BillingTrigger(windowedKey.key(), count,
                    windowedKey.window().startTime(),
                    windowedKey.window().endTime())))
            .to("billing-triggers", Produced.with(Serdes.String(),
                new BillingTriggerSerde()));

        return windowedCounts.map((k, v) -> KeyValue.pair(k.key(), v));
    }
}

A separate billing consumer reads from billing-triggers and calls Stripe for each record:

@Component
public class BillingTriggerConsumer {

    private final StripeClient stripeClient;
    private final PaymentRepository paymentRepository;

    @KafkaListener(topics = "billing-triggers",
                   groupId = "billing-consumer-group")
    @Transactional
    public void processBillingTrigger(
            @Payload BillingTrigger trigger,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset) {

        // UUID generated fresh per record — this is the bug
        String idempotencyKey = UUID.randomUUID().toString();

        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount(trigger.getUnitCount() * 10L) // $0.10 per unit
            .setCurrency("usd")
            .setCustomer(trigger.getCustomerId())
            .putMetadata("billingPeriodStart",
                trigger.getWindowStart().toString())
            .putMetadata("billingPeriodEnd",
                trigger.getWindowEnd().toString())
            .build();

        RequestOptions options = RequestOptions.builder()
            .setIdempotencyKey(idempotencyKey)
            .build();

        Charge charge = stripeClient.charges().create(params, options);

        paymentRepository.save(new PaymentRecord(
            trigger.getCustomerId(),
            trigger.getWindowStart(),
            charge.getId(),
            idempotencyKey));
    }
}

The Kafka Streams application is configured with at-least-once processing (the default — processing.guarantee is not set to exactly_once_v2). The state store daily-usage-store uses the default RocksDB backend with a changelog topic for recovery.

Why this creates a double charge

The developer’s mental model is that suppress(untilWindowCloses()) acts as a deduplication gate at the topology level: each window record is emitted exactly once to billing-triggers, regardless of what happens to the Streams application afterward. This model is correct for a healthy, continuously-running topology. It breaks down during crash-restore cycles without exactly-once processing enabled.

Here is the sequence of events that produces a double charge:

1. The Kafka Streams task processes usage events for customer cust_123 on day 1. The daily window accumulates 450 usage units in the daily-usage-store RocksDB state store.
2. A usage event with a timestamp of 2026-10-02T00:01:00Z (day 2) arrives. Its timestamp advances the stream time past the day-1 window end (2026-10-02T00:00:00Z). The suppress() operator determines that the day-1 window has closed and emits a record with key cust_123 (windowed to day 1) and value 450.
3. The emitted record is written to the billing-triggers Kafka topic (as part of a Kafka producer send, but NOT within a transaction because exactly-once is not enabled).
4. The Kafka Streams task checkpoints the RocksDB state store position and commits the source partition offsets. The checkpoint records the source offset as N (the offset of the day-2 record that advanced the watermark).
5. The billing consumer reads the day-1 record from billing-triggers, generates UUID_A = "3f7c...", and calls Stripe. Stripe commits ch_A. The consumer commits offset M for the billing-triggers partition.
6. The Kafka Streams application crashes (JVM kill, container OOM, Kubernetes pod eviction). The RocksDB checkpoint on disk records offset N-5 — five records behind the actual committed state. The checkpoint lags because RocksDB checkpoints are written asynchronously and the last checkpoint was taken before those five records were processed.
7. On restart, the Streams runtime recovers the state store. It reads the checkpoint file and finds offset N-5. It replays the changelog topic for the daily-usage-store from N-5 forward to rebuild the state store to its current committed state.
8. During replay, the five source records from positions N-4 through N are re-processed against the state store. One of those records is the day-2 timestamp record that originally advanced the watermark past the day-1 window end.
9. The suppress() operator sees the day-1 window close again — the watermark logic is driven by the records being re-processed, and the day-2 timestamp record is one of them. Suppress fires a second time and writes a second record for cust_123 day-1 window to billing-triggers.
10. The billing consumer’s last committed offset for billing-triggers is M. The duplicate record is written at offset M+1. The consumer processes it, generates UUID_B = "9a2e...", and calls Stripe. Stripe has no record of 9a2e... in its idempotency cache. Stripe creates ch_B. Customer cust_123 has been charged twice for day 1: ch_A ($45.00) and ch_B ($45.00).

The failure is not in the suppress() operator itself — it is doing exactly what it is designed to do: emit once per closed window per processing session. The problem is that “once per processing session” means once per uninterrupted run of the topology, not once globally. Each crash-restore cycle is a new processing session, and suppress() will re-emit for any window whose closing event falls within the re-processed range.

The developer misconception

The developer believes that suppress(Suppressed.untilWindowCloses()) provides globally idempotent emission — that a window record, once emitted to the output topic, will never be emitted again regardless of crashes or restarts. This belief is reasonable because the Kafka Streams documentation describes suppress() as emitting “at most one record per window per key”. That guarantee holds within a single execution session. It does not hold across crash boundaries without exactly-once processing enabled, because there is no coordination between the suppress() in-memory watermark state (which is lost on crash and rebuilt from the changelog) and the output topic records that were already written before the crash.

A subtler misconception: the developer assumes that the Kafka producer send of the billing-triggers record and the state-store checkpoint happen atomically. In at-least-once mode they do not. The producer send happens when suppress() fires; the checkpoint happens on the next commit interval (default 30 seconds). Between the send and the checkpoint, the application can crash, leaving the checkpoint pointing to a position before the send. On restore, the send happens again.

The fix

There are two complementary fixes: one at the Kafka Streams topology level and one at the billing consumer level.

At the topology level, enable exactly-once processing so that the billing-triggers producer write and the source partition offset commit are atomic:

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration streamsConfig() {
    Map<String, Object> props = new HashMap<>();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "billing-topology");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
        StreamsConfig.EXACTLY_ONCE_V2); // atomic: producer send + offset commit
    props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
    return new KafkaStreamsConfiguration(props);
}

With EXACTLY_ONCE_V2, the Kafka Streams runtime wraps producer sends and consumer offset commits in a single Kafka transaction. If the task crashes after the transaction is committed, the source partition offset and the billing-triggers write are both durable. On restore, the re-processed range starts after the committed transaction, so the day-2 watermark record is not re-processed, and suppress() does not fire again.

At the billing consumer level, derive the idempotency key from stable business inputs rather than UUID.randomUUID():

@KafkaListener(topics = "billing-triggers",
               groupId = "billing-consumer-group")
@Transactional
public void processBillingTrigger(@Payload BillingTrigger trigger) {

    // Content-hash key: same for any duplicate record for the same
    // customer + billing period. Stripe returns ch_A on the second call.
    String idempotencyKey = computeIdempotencyKey(
        trigger.getCustomerId(),
        trigger.getWindowStart(),
        trigger.getWindowEnd());

    // Check if already processed (optional defence-in-depth)
    if (paymentRepository.existsByCustomerIdAndWindowStart(
            trigger.getCustomerId(), trigger.getWindowStart())) {
        return; // already charged — skip
    }

    ChargeCreateParams params = ChargeCreateParams.builder()
        .setAmount(trigger.getUnitCount() * 10L)
        .setCurrency("usd")
        .setCustomer(trigger.getCustomerId())
        .putMetadata("billingPeriodStart",
            trigger.getWindowStart().toString())
        .build();

    RequestOptions options = RequestOptions.builder()
        .setIdempotencyKey(idempotencyKey) // stable key
        .build();

    Charge charge = stripeClient.charges().create(params, options);
    paymentRepository.save(new PaymentRecord(
        trigger.getCustomerId(), trigger.getWindowStart(),
        charge.getId(), idempotencyKey));
}

private String computeIdempotencyKey(
        String customerId, Instant windowStart, Instant windowEnd) {
    String input = customerId + ":" +
                   windowStart.getEpochSecond() + ":" +
                   windowEnd.getEpochSecond() + ":billing";
    try {
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        byte[] hash = md.digest(input.getBytes(StandardCharsets.UTF_8));
        return HexFormat.of().formatHex(hash).substring(0, 32);
    } catch (NoSuchAlgorithmException e) {
        throw new IllegalStateException(e);
    }
}

The content-hash key is identical for any duplicate record representing the same customer and billing window. Stripe’s 24-hour idempotency cache returns ch_A on the second call, preventing ch_B. The optional database check provides defence-in-depth against duplicates that arrive after the 24-hour Stripe cache TTL.

Mode 2: EXACTLY_ONCE_V2 with a .foreach() Stripe call — the Kafka transaction covers only internal topic writes and offset commits; the external HTTP call happens outside the transaction boundary; a producer epoch fence causes the Streams runtime to re-execute the step and call Stripe twice

The setup

The developer migrates the billing topology to use EXACTLY_ONCE_V2 after reading about the state-store restore issue described above. They enable EXACTLY_ONCE_V2 in the streams configuration and add a .foreach() at the end of the topology to call Stripe directly from within the topology, reasoning that the exactly-once guarantee will prevent duplicate calls:

@Bean
public KStream<String, BillingTrigger> buildBillingTopology(
        StreamsBuilder builder) {

    KTable<Windowed<String>, Long> dailyUsage = builder
        .stream("usage-events",
            Consumed.with(Serdes.String(), Serdes.Long()))
        .groupByKey()
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofDays(1)))
        .count(Materialized.as("daily-usage-store"));

    dailyUsage
        .suppress(Suppressed.untilWindowCloses(
            BufferConfig.unbounded()))
        .toStream()
        .foreach((windowedKey, count) -> {
            // Developer believes EXACTLY_ONCE_V2 covers this call
            String idempotencyKey = UUID.randomUUID().toString();

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(count * 10L)
                .setCurrency("usd")
                .setCustomer(windowedKey.key())
                .build();

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build();

            try {
                Charge charge = stripeClient.charges()
                    .create(params, options);
                log.info("Charged {} for {} units: {}",
                    windowedKey.key(), count, charge.getId());
            } catch (StripeException e) {
                throw new RuntimeException("Stripe charge failed", e);
            }
        });

    return builder.stream("usage-events",
        Consumed.with(Serdes.String(), Serdes.Long()))
        .mapValues(v -> (BillingTrigger) null); // placeholder return
}

The streams configuration enables exactly-once:

props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
    StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);

The developer removes the separate billing-triggers Kafka topic and the billing consumer. The Stripe call now happens inline in the topology via .foreach().

Why this creates a double charge

The EXACTLY_ONCE_V2 guarantee in Kafka Streams is specifically about the atomicity of Kafka producer sends and consumer offset commits. The runtime wraps these operations in a Kafka transaction: at the end of each commit interval, it atomically writes all pending output records and commits the source partition offsets within a single transaction. If the transaction succeeds, both the writes and the offset commits are durable. If the transaction is aborted, neither is durable and the processing step is retried from the last committed offset.

The critical boundary: the Kafka transaction does not and cannot include external HTTP calls. .foreach() is a terminal operator that executes user-provided code for each record as the topology processes it. This code runs during the Streams processing loop, before the commit interval fires and before the Kafka transaction is opened or committed. The sequence is:

1. Streams processing loop runs. For a record representing customer cust_123 day-1 window closing, the .foreach() lambda is invoked.
2. The lambda generates UUID_A and calls Stripe. Stripe commits ch_A. The lambda returns.
3. The commit interval fires. The Streams runtime opens a Kafka transaction, writes pending output records to internal topics, and prepares to commit consumer offsets.
4. At this moment, a transient network partition causes the broker to elect a new partition leader. The previous leader had fenced the producer epoch. The Streams runtime’s producer receives a ProducerFencedException when it attempts to commit the transaction.
5. The Streams runtime catches the exception, marks the task as failed, and resets to the last committed source partition offset (which pre-dates step 1).
6. On the next processing cycle, the Streams runtime re-reads the source records from the last committed offset. The record that triggered the .foreach() lambda for cust_123 day-1 window is re-read.
7. The .foreach() lambda is invoked again. It generates UUID_B = "9a2e...". It calls Stripe. Stripe has no record of 9a2e... — it creates ch_B.
8. The Kafka transaction this time commits successfully. The source partition offset is committed past the triggering record.
9. Customer cust_123 has been charged twice: ch_A and ch_B.

The same failure occurs in any scenario that causes the Streams runtime to re-process a record from before the last committed offset: producer epoch fence (as above), task migration to a different instance during a rebalance, or a Streams thread crash after the .foreach() lambda runs but before the transaction commits.

The developer misconception

The developer believes that EXACTLY_ONCE_V2 means every record in the topology is processed exactly once, including all user code in terminal operators like .foreach(), .peek(), and .process() when those operators contain external side effects. This is a reasonable but incorrect extrapolation from the name of the guarantee.

The EXACTLY_ONCE_V2 guarantee is scoped to the Kafka message system: each input record produces its corresponding output records and offset commits exactly once in the Kafka log. It says nothing about user code that runs as a side effect of processing. From the Kafka Streams perspective, the external HTTP call to Stripe is invisible — the runtime has no way to include it in a Kafka transaction, and no way to know whether it succeeded before a crash.

A subtler misconception: the developer may believe that because they wrap the Streams configuration with EXACTLY_ONCE_V2, the ProducerFencedException scenario (which requires a zombie producer to be fenced by a new producer after an epoch increment) cannot happen to them because exactly-once prevents zombie producers. In fact, the opposite is true: exactly-once mode uses epoch-based producer fencing as its primary mechanism. The fencing is what causes the ProducerFencedException — it is a normal operating event in an exactly-once Kafka Streams application, not an error condition. The transaction abort and retry is how EXACTLY_ONCE_V2 recovers from fencing. External side effects in .foreach() are always exposed to this retry.

The fix

Move the Stripe call out of .foreach() and into a downstream Kafka consumer that uses a stable idempotency key. The topology should write a billing trigger record to a Kafka topic (inside the Kafka transaction), and a separate process consumes that topic and calls Stripe:

// In the topology — write to Kafka topic (inside the transaction)
dailyUsage
    .suppress(Suppressed.untilWindowCloses(
        BufferConfig.unbounded()))
    .toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new BillingTrigger(
            windowedKey.key(), count,
            windowedKey.window().startTime(),
            windowedKey.window().endTime())))
    .to("billing-triggers",
        Produced.with(Serdes.String(), new BillingTriggerSerde()));
    // This write is inside the EXACTLY_ONCE_V2 Kafka transaction.
    // If the transaction aborts, this write is rolled back and
    // the record will not appear on billing-triggers.
    // The billing consumer never sees a duplicate billing trigger.

// In the billing consumer — stable idempotency key
@KafkaListener(topics = "billing-triggers",
               groupId = "billing-consumer-group")
public void processBillingTrigger(@Payload BillingTrigger trigger) {

    // Stable key: same for any invocation for the same
    // customer + billing window
    String idempotencyKey = computeIdempotencyKey(
        trigger.getCustomerId(),
        trigger.getWindowStart(),
        trigger.getWindowEnd());

    ChargeCreateParams params = ChargeCreateParams.builder()
        .setAmount(trigger.getUnitCount() * 10L)
        .setCurrency("usd")
        .setCustomer(trigger.getCustomerId())
        .build();

    RequestOptions options = RequestOptions.builder()
        .setIdempotencyKey(idempotencyKey)
        .build();

    Charge charge = stripeClient.charges().create(params, options);
    // ... persist
}

With this architecture, the EXACTLY_ONCE_V2 guarantee is applied where it is effective: ensuring that each billing trigger record appears exactly once on the billing-triggers topic. The billing consumer, which operates outside the Kafka transaction, uses a content-hash idempotency key to ensure that even if the consumer re-processes a record (due to its own consumer offset commit failure), the Stripe call is idempotent.

If the use case requires inline Stripe calls from the topology (for example, because a downstream Kafka topic is not acceptable), the fix is still to use a stable idempotency key in the .foreach() lambda:

.foreach((windowedKey, count) -> {
    // Stable key — same UUID for same customer + same window period
    String idempotencyKey = computeIdempotencyKey(
        windowedKey.key(),
        windowedKey.window().startTime(),
        windowedKey.window().endTime());

    ChargeCreateParams params = ChargeCreateParams.builder()
        .setAmount(count * 10L)
        .setCurrency("usd")
        .setCustomer(windowedKey.key())
        .build();

    RequestOptions options = RequestOptions.builder()
        .setIdempotencyKey(idempotencyKey) // stable — same on retry
        .build();

    Charge charge = stripeClient.charges().create(params, options);
    // On the retry after transaction abort, Stripe returns ch_A
    // because the idempotency key is the same — ch_B is never created.
});

The inline fix works because Stripe’s 24-hour idempotency cache uses the Idempotency-Key header value as the cache key. If the second invocation (after transaction abort and retry) sends the same key, Stripe returns the cached response from the first invocation. The stable content-hash key ensures the two invocations are indistinguishable from Stripe’s perspective.

Mode 3: PunctuationType.STREAM_TIME billing punctuator fires multiple times when a batch of records spans multiple billing periods — three separate Stripe charges created for what should be one

The setup

The developer builds a billing processor using the Kafka Streams Processor API, using a STREAM_TIME-based punctuator to fire billing once per day. The processor accumulates usage counts in a state store and charges each customer when the punctuator fires:

public class DailyBillingProcessor
        implements Processor<String, Long, Void, Void> {

    private ProcessorContext<Void, Void> context;
    private KeyValueStore<String, Long> usageStore;
    private final StripeClient stripeClient;
    private final PaymentRepository paymentRepository;

    @Override
    public void init(ProcessorContext<Void, Void> context) {
        this.context = context;
        this.usageStore = context.getStateStore("usage-accumulator");

        // Schedule daily billing at stream-time intervals
        context.schedule(
            Duration.ofDays(1),
            PunctuationType.STREAM_TIME,
            this::chargeAllCustomers
        );
    }

    @Override
    public void process(Record<String, Long> record) {
        String customerId = record.key();
        Long currentUsage = usageStore.get(customerId);
        long newUsage = (currentUsage == null ? 0L : currentUsage)
                      + record.value();
        usageStore.put(customerId, newUsage);
    }

    private void chargeAllCustomers(long punctuationTimestamp) {
        KeyValueIterator<String, Long> iter = usageStore.all();
        while (iter.hasNext()) {
            KeyValue<String, Long> kv = iter.next();
            if (kv.value > 0) {
                // UUID generated fresh per punctuation per customer
                String idempotencyKey = UUID.randomUUID().toString();

                ChargeCreateParams params = ChargeCreateParams.builder()
                    .setAmount(kv.value * 10L)
                    .setCurrency("usd")
                    .setCustomer(kv.key)
                    .putMetadata("billingTimestamp",
                        String.valueOf(punctuationTimestamp))
                    .build();

                RequestOptions options = RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build();

                try {
                    Charge charge = stripeClient.charges()
                        .create(params, options);
                    usageStore.put(kv.key, 0L); // reset after charging
                    paymentRepository.save(new PaymentRecord(
                        kv.key, punctuationTimestamp,
                        charge.getId(), idempotencyKey));
                } catch (StripeException e) {
                    log.error("Stripe charge failed for {}", kv.key, e);
                }
            }
        }
        iter.close();
    }

    @Override
    public void close() {}
}

The topology is built using the Processor API integration with the DSL:

@Bean
public KStream<String, Long> buildBillingTopology(
        StreamsBuilder builder) {

    StoreBuilder<KeyValueStore<String, Long>> storeBuilder =
        Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore("usage-accumulator"),
            Serdes.String(),
            Serdes.Long());

    builder.addStateStore(storeBuilder);

    return builder
        .stream("usage-events",
            Consumed.with(Serdes.String(), Serdes.Long()))
        .process(
            () -> new DailyBillingProcessor(stripeClient, paymentRepository),
            "usage-accumulator");
}

Why this creates three charges

The failure arises from a gap in the event stream. Consider the following scenario:

The Kafka broker undergoes planned maintenance over a long weekend. The usage-events topic producers continue writing usage events throughout the weekend, but the Kafka Streams application is down. Events accumulate with timestamps spanning Friday October 10 at 15:00 UTC through Monday October 13 at 09:00 UTC — approximately 66 hours of events.

On Monday morning, the Kafka Streams application restarts and begins processing the backlogged events. The first Kafka poll() call returns a batch of records from across the three-day gap. Here is what happens:

1. The Streams runtime polls the usage-events topic and receives a batch of 10,000 records with timestamps ranging from Friday 15:00 to Monday 09:00.
2. Processing begins. The first record has timestamp 2026-10-10T15:00:00Z (stream time starts here). Each subsequent record advances the stream time to the maximum timestamp seen so far.
3. The Streams runtime processes records. The state store accumulates usage for each customer across all three days of events.
4. Stream time advances past 2026-10-11T15:00:00Z (Friday + 24 hours). The STREAM_TIME punctuator fires: chargeAllCustomers(2026-10-11T15:00:00Z).
5. The punctuator iterates all customers in the state store. At this point, the store contains the accumulated usage from Friday 15:00 through the current stream-time position (Saturday 15:00). For customer cust_123, the accumulated count is 300. A Stripe charge is created with UUID_A: ch_A = $30.00. The state store resets cust_123 to 0.
6. Processing continues. More records are processed. Stream time advances past 2026-10-12T15:00:00Z (Saturday + 24 hours). The punctuator fires again: chargeAllCustomers(2026-10-12T15:00:00Z).
7. For customer cust_123, the accumulated count since the last reset (Saturday 15:00 through Sunday 15:00) is 280. A Stripe charge is created with UUID_B: ch_B = $28.00.
8. Processing continues. Stream time advances past 2026-10-13T09:00:00Z. But 2026-10-13T15:00:00Z has not been reached (the latest event timestamp is 09:00). The punctuator does not fire a third time during this batch.

The result: two Stripe charges (ch_A and ch_B) are created during a single processing run, covering two distinct billing periods from the backlog. In a scenario where the gap spans three full 24-hour intervals (e.g., a 73-hour outage starting at 15:00 on Thursday), the punctuator fires three times: Thursday→Friday, Friday→Saturday, Saturday→Sunday. Customer cust_123 receives three charges during one processing catch-up run.

The charges themselves are not “double charges” in the traditional sense — each punctuation charge covers a distinct time window and a non-overlapping usage count (because the state store is reset after each punctuation). The problem is that the developer expected the system to produce one billing event for the entire backlog period and is surprised to receive three invoices.

However, a true double-charge variant exists when the state-store reset and the Stripe call are not atomic. If the Streams task crashes after calling Stripe (step 5) but before committing the state-store reset (usageStore.put(kv.key, 0L) and the downstream processing), and the source partition offset commit also fails (because the Streams commit interval fires after the Stripe call but the task crashes before the commit completes), then on recovery, the punctuator fires again for the same stream-time interval, with the same non-zero usage count in the state store, and generates UUID_C — a true duplicate charge for the same billing period.

The developer misconception

The developer believes that PunctuationType.STREAM_TIME fires “once per day of real event data processed” — that a 66-hour batch produces one punctuation at the end, not three punctuations at each 24-hour boundary within the batch. This belief is reinforced by the name “STREAM_TIME” — it sounds like it tracks the progression of the data stream, not the wall-clock time.

The actual semantics: stream time is the maximum timestamp across all records seen so far on the partition. Every time stream time advances past a multiple of the schedule interval, the punctuator fires. If stream time jumps from T to T+66h in a single processing batch (because 66 hours of records are processed in one poll loop), the 24-hour punctuator fires at T+24h, T+48h, and T+72h (if reached) — three times in that single processing loop, one per elapsed 24-hour interval.

A related misconception: the developer assumes that STREAM_TIME always fires at most once per commit interval. This is also incorrect. Multiple punctuations can fire within a single processing loop if the stream time advances past multiple schedule boundaries during that loop. The commit interval controls how often Kafka offsets and state stores are checkpointed; it does not control how often STREAM_TIME punctuators fire.

The fix

The fix has two parts: use a stable idempotency key tied to the billing period, and understand the punctuation semantics for your use case.

For the idempotency key, derive it from the punctuationTimestamp and the customer ID. The timestamp passed to the punctuator is the stream time at the moment it fires, which is deterministic for a given input sequence:

private void chargeAllCustomers(long punctuationTimestamp) {
    // Align to day boundary so that crash-retry produces the same key
    Instant billingDay = Instant.ofEpochMilli(punctuationTimestamp)
        .truncatedTo(ChronoUnit.DAYS);

    KeyValueIterator<String, Long> iter = usageStore.all();
    while (iter.hasNext()) {
        KeyValue<String, Long> kv = iter.next();
        if (kv.value > 0) {
            // Stable key: same customer + same billing day = same key
            String idempotencyKey = computeIdempotencyKey(
                kv.key, billingDay);

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(kv.value * 10L)
                .setCurrency("usd")
                .setCustomer(kv.key)
                .putMetadata("billingDay", billingDay.toString())
                .build();

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey) // stable per day
                .build();

            try {
                Charge charge = stripeClient.charges()
                    .create(params, options);
                usageStore.put(kv.key, 0L);
                paymentRepository.save(new PaymentRecord(
                    kv.key, billingDay, charge.getId(), idempotencyKey));
            } catch (StripeException e) {
                log.error("Stripe charge failed for {}", kv.key, e);
            }
        }
    }
    iter.close();
}

private String computeIdempotencyKey(String customerId, Instant billingDay) {
    String input = customerId + ":" +
                   billingDay.getEpochSecond() + ":daily-billing";
    try {
        MessageDigest md = MessageDigest.getInstance("SHA-256");
        byte[] hash = md.digest(input.getBytes(StandardCharsets.UTF_8));
        return HexFormat.of().formatHex(hash).substring(0, 32);
    } catch (NoSuchAlgorithmException e) {
        throw new IllegalStateException(e);
    }
}

With this fix, the crash-retry scenario no longer creates a duplicate charge. On the retry, chargeAllCustomers fires with the same punctuationTimestamp (because stream time replays to the same position), computes the same stable key, and Stripe returns ch_A from its idempotency cache.

For the multiple-punctuation scenario (three charges for a 66-hour backlog), the fix depends on the intended billing semantics. If the business requirement is “bill once per calendar day regardless of when the batch is processed”, then three charges for a 66-hour backlog is the correct behavior — it just needs to be communicated clearly in the customer invoice. If the business requirement is “bill once for the entire backlog period regardless of its length”, then STREAM_TIME punctuation is the wrong tool. Instead, use a single billing cycle driven by wall-clock time or by a post-processing aggregation step:

// Alternative: use WALL_CLOCK_TIME if billing should fire
// at most once per calendar day regardless of stream-time gaps.
// WALL_CLOCK_TIME fires based on actual elapsed real-world time,
// not the timestamps in the event data.
context.schedule(
    Duration.ofDays(1),
    PunctuationType.WALL_CLOCK_TIME, // fires once per real day
    this::chargeAllCustomers
);

With PunctuationType.WALL_CLOCK_TIME, a 66-hour processing catch-up run that completes in 10 minutes fires the punctuator 0 times during catch-up (because only 10 real minutes elapsed) and once on the following real-time day. All accumulated usage from the backlog is billed in that single punctuation. The trade-off is that WALL_CLOCK_TIME has different behavior during sustained outages and requires careful consideration of what “accumulate-then-charge” means if the application is down for multiple real days.

Integration test patterns for all three modes

Testing these failure modes requires simulating specific Kafka Streams runtime conditions. The following patterns use TopologyTestDriver for deterministic control over stream time and record processing, and WireMock with Testcontainers for realistic Stripe API simulation.

Testing Mode 1: suppress() re-emission via state-store replay simulation

@Test
void suppressEmitsDuplicateAfterCrashRestore() {
    // Build topology with at-least-once processing
    Topology topology = buildBillingTopology();
    TopologyTestDriver driver = new TopologyTestDriver(topology,
        buildStreamsConfig(StreamsConfig.AT_LEAST_ONCE));

    TestInputTopic<String, Long> input = driver.createInputTopic(
        "usage-events", new StringSerializer(), new LongSerializer());
    TestOutputTopic<String, BillingTrigger> output =
        driver.createOutputTopic("billing-triggers",
            new StringDeserializer(), new BillingTriggerDeserializer());

    // Process day-1 events at stream time within day 1
    Instant day1Start = Instant.parse("2026-10-10T00:00:00Z");
    input.pipeInput("cust_123", 150L,
        day1Start.plusSeconds(3600).toEpochMilli());  // 01:00
    input.pipeInput("cust_123", 300L,
        day1Start.plusSeconds(7200).toEpochMilli());  // 02:00

    // Advance stream time past day-1 window end to trigger suppress
    Instant day2Start = Instant.parse("2026-10-11T00:00:00Z");
    input.pipeInput("cust_456", 1L,
        day2Start.plusSeconds(60).toEpochMilli());    // day 2, 00:01

    // suppress() should emit the day-1 record
    assertThat(output.getQueueSize()).isEqualTo(1);
    BillingTrigger trigger1 = output.readValue();
    assertThat(trigger1.getCustomerId()).isEqualTo("cust_123");
    assertThat(trigger1.getUnitCount()).isEqualTo(450L);

    // Simulate crash by closing and reopening the driver
    // (In production, this models the state-store being replayed
    // past the checkpoint — the suppress watermark resets)
    driver.close();

    // Reopen from the beginning (simulating changelog replay
    // that re-processes the day-2 record that advanced the watermark)
    TopologyTestDriver driver2 = new TopologyTestDriver(topology,
        buildStreamsConfig(StreamsConfig.AT_LEAST_ONCE));

    TestInputTopic<String, Long> input2 = driver2.createInputTopic(
        "usage-events", new StringSerializer(), new LongSerializer());
    TestOutputTopic<String, BillingTrigger> output2 =
        driver2.createOutputTopic("billing-triggers",
            new StringDeserializer(), new BillingTriggerDeserializer());

    // Re-process the day-2 record (simulating changelog replay)
    input2.pipeInput("cust_456", 1L,
        day2Start.plusSeconds(60).toEpochMilli());

    // suppress() fires again — duplicate emission
    assertThat(output2.getQueueSize()).isEqualTo(1);
    BillingTrigger trigger2 = output2.readValue();
    assertThat(trigger2.getCustomerId()).isEqualTo("cust_123");
    assertThat(trigger2.getUnitCount()).isEqualTo(450L);

    // The billing consumer must handle this duplicate with
    // a stable idempotency key — not UUID.randomUUID()
    driver2.close();
}

Testing Mode 2: EXACTLY_ONCE_V2 .foreach() re-execution on transaction abort

@ExtendWith(MockitoExtension.class)
class ExactlyOnceForEachTest {

    @Mock
    private StripeClient stripeClient;

    @Test
    void foreachCalledTwiceOnTransactionAbortRetry() throws Exception {
        // Capture all idempotency keys sent to Stripe
        List<String> capturedKeys = new ArrayList<>();
        ArgumentCaptor<RequestOptions> optionsCaptor =
            ArgumentCaptor.forClass(RequestOptions.class);

        Charge mockCharge = mock(Charge.class);
        when(mockCharge.getId()).thenReturn("ch_mock");
        when(stripeClient.charges().create(any(), optionsCaptor.capture()))
            .thenReturn(mockCharge);

        // Build topology with .foreach() calling Stripe
        Topology topology = new BillingTopologyWithForeach(stripeClient)
            .build();

        TopologyTestDriver driver = new TopologyTestDriver(topology,
            buildStreamsConfig(StreamsConfig.EXACTLY_ONCE_V2));

        TestInputTopic<String, Long> input = driver.createInputTopic(
            "usage-events", new StringSerializer(), new LongSerializer());

        // Process a record that triggers the daily window close
        Instant day1Start = Instant.parse("2026-10-10T00:00:00Z");
        input.pipeInput("cust_123", 450L,
            day1Start.plusSeconds(3600).toEpochMilli());

        // Advance past window end
        Instant day2Start = Instant.parse("2026-10-11T00:00:00Z");
        input.pipeInput("cust_123", 1L,
            day2Start.plusSeconds(60).toEpochMilli());

        // .foreach() fired once — one Stripe call
        assertThat(stripeClient.charges().create(any(), any()))
            .wasCalledOnce();
        String firstKey = optionsCaptor.getValue().getIdempotencyKey();

        // Simulate re-processing (what happens after transaction abort)
        // by re-injecting the same record
        input.pipeInput("cust_123", 450L,
            day1Start.plusSeconds(3600).toEpochMilli());
        input.pipeInput("cust_123", 1L,
            day2Start.plusSeconds(60).toEpochMilli());

        // .foreach() fires again — second Stripe call
        List<String> allKeys = optionsCaptor.getAllValues();
        assertThat(allKeys).hasSize(2);

        // With UUID.randomUUID(): the two keys differ — ch_B would be created
        assertThat(allKeys.get(0)).isNotEqualTo(allKeys.get(1));
        // Demonstrate the problem: two different keys = two Stripe charges

        // With content-hash key: the two keys must be equal
        // Stripe's idempotency cache returns ch_A on the second call
        String expectedKey = computeIdempotencyKey(
            "cust_123",
            day1Start,
            day2Start);
        assertThat(allKeys.get(0)).isEqualTo(expectedKey);
        assertThat(allKeys.get(1)).isEqualTo(expectedKey);

        driver.close();
    }
}

Testing Mode 3: STREAM_TIME punctuator firing multiple times on timestamp gap

@Test
void streamTimePunctuatorFiresMultipleTimesOnTimestampGap() {
    // Capture punctuation timestamps and Stripe charges
    List<Long> punctuationTimestamps = new ArrayList<>();
    List<String> chargeIdempotencyKeys = new ArrayList<>();

    Topology topology = buildBillingProcessorTopology(
        punctuationTimestamps, chargeIdempotencyKeys);

    TopologyTestDriver driver = new TopologyTestDriver(topology,
        buildStreamsConfig(StreamsConfig.AT_LEAST_ONCE));

    TestInputTopic<String, Long> input = driver.createInputTopic(
        "usage-events", new StringSerializer(), new LongSerializer());

    // Simulate Friday 15:00 — system starts, accumulate usage
    Instant fri15 = Instant.parse("2026-10-10T15:00:00Z");
    input.pipeInput("cust_123", 100L, fri15.toEpochMilli());
    input.pipeInput("cust_123", 50L,
        fri15.plusSeconds(3600).toEpochMilli()); // Fri 16:00

    // Skip weekend — jump to Monday 09:00 (66h gap)
    Instant mon09 = Instant.parse("2026-10-13T09:00:00Z");
    input.pipeInput("cust_123", 75L, mon09.toEpochMilli());

    // Assert: STREAM_TIME punctuator fired TWICE
    // (at Fri+24h = Sat 15:00, and at Sat+24h = Sun 15:00)
    // Mon+24h = Tue 09:00 not reached because last event is Mon 09:00
    assertThat(punctuationTimestamps).hasSize(2);
    assertThat(punctuationTimestamps.get(0))
        .isEqualTo(Instant.parse("2026-10-11T15:00:00Z").toEpochMilli());
    assertThat(punctuationTimestamps.get(1))
        .isEqualTo(Instant.parse("2026-10-12T15:00:00Z").toEpochMilli());

    // Assert: two Stripe charges created in one processing run
    assertThat(chargeIdempotencyKeys).hasSize(2);

    // With stable key: both keys are based on billing day, not UUID
    // Sat 15:00 punctuation → billing day = Sat (truncated to day)
    // Sun 15:00 punctuation → billing day = Sun
    String expectedKey1 = computeIdempotencyKey("cust_123",
        Instant.parse("2026-10-11T00:00:00Z")); // Sat
    String expectedKey2 = computeIdempotencyKey("cust_123",
        Instant.parse("2026-10-12T00:00:00Z")); // Sun

    assertThat(chargeIdempotencyKeys.get(0)).isEqualTo(expectedKey1);
    assertThat(chargeIdempotencyKeys.get(1)).isEqualTo(expectedKey2);

    driver.close();
}

How the three failure modes differ from the rest of this series

The Kafka Streams DSL failure modes differ from the REST controller, reactive, and scheduler failure modes covered in earlier posts in one important way: the point of re-execution is determined by the Streams runtime, not by user code or infrastructure configuration. In the Spring WebMVC, Feign, or OkHttp posts, re-execution is triggered by user-configured retry policies or HTTP client timeout behavior. In Kafka Streams, re-execution is triggered by the Streams runtime’s crash-recovery mechanism, transaction abort handling, and punctuation scheduling — all of which are internal to the runtime and operate without the developer’s direct involvement.

This distinction matters for debugging. When a double charge appears in the Stripe dashboard after a Feign client retry, the developer can trace it to a specific HTTP client configuration or timeout event. When a double charge appears after a Kafka Streams task crash, the developer must understand the state-store checkpoint mechanism, the exactly-once transaction scope, and the stream-time punctuation semantics to identify which execution path produced the duplicate. The root cause is always the same — a UUID generated fresh on re-execution — but the diagnosis requires Kafka Streams-specific knowledge.

Comparison table

The invariant that prevents all three failures

All three modes share the same root cause: UUID.randomUUID() generates a different value on each invocation, and re-invocation is a normal operating condition in Kafka Streams — not an error. The Streams runtime is designed to re-process records after crash-recovery, transaction abort, and state-store restore. Re-execution is a feature, not a bug. The bug is assuming that re-execution will not happen.

The invariant that prevents all three failures: the Stripe Idempotency-Key must be a deterministic function of the stable business inputs that define the logical charge. For daily usage-based billing, the stable inputs are the customer ID and the billing day. The key sha256("cust_123:2026-10-10:daily-billing")[:32] is identical regardless of which Streams task processes the record, which JVM instance runs the billing consumer, whether the processing is happening for the first time or the fifteenth time after fifteen consecutive crash-restores, and whether the punctuator fired once or three times during a single catch-up run. Stripe’s 24-hour idempotency cache uses this key to deduplicate — every call with the same key returns the same response, and only one charge is ever committed.

The content-hash approach also makes the failure modes auditable: the idempotency_key column in the payment_records table contains a value derived from the billing inputs, not from UUID.randomUUID(). When a potential double-charge is reported, you can compute the expected key for any customer and billing day and verify in the Stripe dashboard whether that key appears once or twice. With UUID-based keys, auditing requires matching timestamps across the Stripe dashboard, the application logs, and the Kafka consumer offset history — a much harder correlation.