Apache Kafka Streams and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Apache Kafka Streams is a Java library for building stream processing applications directly on Kafka. Unlike a plain Kafka consumer, Kafka Streams manages local RocksDB state stores, runs its own commit cycle, and provides a processing.guarantee setting that many teams interpret as “exactly-once end-to-end.” When Kafka Streams drives billing — a usage event enters the topology, a Processor calls stripe.charges.create(), and the result updates a state store — three Kafka Streams-specific behaviors introduce Stripe double-charge failure modes that neither Kafka consumer lag metrics nor the exactly_once_v2 setting prevents.

This post covers those three failure modes with Java Kafka Streams code, content-hash idempotency keys stable across task migrations and repartition operations, per-billing-period vault keys via a spend-cap proxy, and pre-flight database checks — the two-layer governance pattern that closes all three without changing topology structure, partition count, or processing guarantee setting.

Failure mode 1: at_least_once commit cycle flushes state store changelogs before committing consumer offsets — OOM kill or session timeout between them causes task re-assignment starting from the last committed consumer offset

Kafka Streams maintains local RocksDB state stores for stateful operations: KTable lookups, aggregations, windowed joins, and explicit KeyValueStore state attached to Processor implementations. Each state store is backed by a Kafka changelog topic. When a Streams instance writes to a state store, the write goes to the in-memory RocksDB buffer first. The Kafka Streams commit cycle flushes these in-memory buffers to disk (RocksDB compaction) and writes the pending changelog records to Kafka, then commits the consumer group offsets for all input partitions.

The commit cycle order matters: state store changelog records are sent to Kafka before the consumer offset commit. This sequencing exists so that, on restart, a new task owner can restore the state store from the changelog (which includes records up to the changelog flush) and then start reading input events from the committed consumer offset. If the changelog flush and consumer offset commit were reversed, restarted tasks would read ahead of the state they restored — the state store would be stale relative to the input position.

The failure scenario: a billing Processor reads a batch of usage events within one commit interval (commit.interval.ms, default 100 ms in Kafka Streams 3.x). For each event, it calls stripe.charges.create() and updates a billing-status KeyValueStore. At the end of the commit interval, Kafka Streams flushes the state store changelog to Kafka — the KeyValueStore updates for all processed events are now durable in Kafka. Kafka Streams then calls consumer.commitSync() for the input partitions. Between the changelog flush completing and the commitSync() call, the JVM receives an OOM eviction from the Kubernetes scheduler — heap pressure from holding billing event payloads, Stripe response objects, and RocksDB off-heap memory triggers SIGKILL. The consumer group offset in Kafka remains at the position before the batch. The state store changelog in Kafka is up-to-date.

When a new Streams instance takes over the task, it restores the KeyValueStore from its changelog topic: the state store shows every customer in the batch as “billed.” It then begins reading input events from the last committed consumer offset — the start of the batch that was killed mid-commit. Every billing event in the batch is re-delivered to process(). If the billing processor checks the KeyValueStore before calling Stripe (“is this customer already billed for this period?”), the check returns true for every event — the state store correctly reflects what was billed before the crash. The deduplication guard holds.

But the deduplication guard fails when the state store write comes after the Stripe call — the natural order, since you update billing status after confirming the charge:

import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.KeyValueStore;
import com.stripe.Stripe;
import com.stripe.model.Charge;
import com.stripe.param.ChargeCreateParams;

public class BillingProcessor implements Processor<String, BillingEvent, String, BillingResult> {

    private ProcessorContext<String, BillingResult> context;
    private KeyValueStore<String, String> billingStatusStore;

    @Override
    public void init(ProcessorContext<String, BillingResult> context) {
        this.context = context;
        this.billingStatusStore = context.getStateStore("billing-status-store");
    }

    @Override
    public void process(Record<String, BillingEvent> record) {
        String customerId = record.key();
        BillingEvent event = record.value();
        String storeKey = customerId + ":" + event.getBillingPeriod();

        // UNSAFE ordering: check → Stripe call → state store write
        // If OOM kill fires between stripe.charges.create() returning and
        // billingStatusStore.put(), the state store does not have the "billed"
        // entry. On task migration, the state store is restored from changelog
        // (no entry for this customer+period). The consumer offset re-delivers
        // this event to process(). The check below returns null → stripe.charges.create()
        // is called again → ch_B is created alongside ch_A.
        String existingStatus = billingStatusStore.get(storeKey);
        if (existingStatus != null) {
            return; // already billed — skip
        }

        // stripe.charges.create() takes 10-20 seconds under load.
        // ch_A is created on Stripe's servers when this call returns.
        // The state store has not been updated yet.
        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount((long) event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getStripeCustomerId())
            .setDescription("Usage billing for " + event.getBillingPeriod())
            // no idempotency key — re-processing creates ch_B
            .build();
        Charge charge = Charge.create(params);

        // OOM kill between Stripe returning and this put() call:
        // the changelog topic does not get this record.
        // Restored state store shows no entry for this customer+period.
        // Re-processing processor re-charges the customer.
        billingStatusStore.put(storeKey, charge.getId());

        context.forward(record.withValue(new BillingResult(customerId, charge.getId())));
    }
}

The fix for the OOM-kill window is to write to the state store before calling stripe.charges.create() — and to use a content-hash Stripe idempotency key so that if the crash fires between the state store write and the Stripe call, the re-delivered event produces the same idempotency key and Stripe deduplicates it. This is the write-before-Stripe pattern for state stores: the state store write records billing intent before the side effect, and the idempotency key makes the side effect safe to repeat. However, a state store alone is not a sufficient guard once the Stripe idempotency window closes (24 hours), which is why the pre-flight database check against PostgreSQL — durable outside the Kafka commit cycle — is required for long-running consumer group lag scenarios.

Failure mode 2: processing.guarantee=exactly_once_v2 atomically commits Kafka output records and consumer offsets — but stripe.charges.create() executes outside the Kafka transaction boundary

Kafka Streams’ exactly_once_v2 processing guarantee (the default since Kafka 3.0, replacing the deprecated exactly_once and at_least_once defaults in older versions) uses Kafka’s producer transaction API to atomically write output records and commit input consumer offsets. The Streams instance creates a transactional producer, calls producer.beginTransaction() at the start of each commit cycle, writes output records to downstream Kafka topics via the transactional producer, calls producer.sendOffsetsToTransaction(offsets, groupMetadata) to include the consumer offset commit in the transaction, and then calls producer.commitTransaction(). If commitTransaction() fails, Kafka aborts the transaction — output records are rolled back (they are marked with the transaction’s abort marker and invisible to downstream consumers reading with isolation.level=read_committed) and the consumer offset is not advanced. The input events are re-delivered on the next poll.

The exactly_once_v2 guarantee applies only to Kafka-to-Kafka writes. An HTTP call to stripe.charges.create() inside the process() method is a side effect that runs before the Kafka transaction commits. From Kafka Streams’ perspective, the call to stripe.charges.create() is invisible — it is not a Kafka producer write, not a Kafka consumer offset commit, and not a state store changelog write. The sequence is:

1. producer.beginTransaction()
2. process() runs: stripe.charges.create() → ch_A created on Stripe’s servers (HTTP 200)
3. Kafka Streams writes output records to the downstream topic via the transactional producer
4. producer.sendOffsetsToTransaction(...) — consumer offset included in transaction
5. producer.commitTransaction() → transaction commit fails: the Kafka broker elected a new partition leader mid-commit; the previous leader’s epoch was bumped; the transactional producer’s epoch is now fenced; ProducerFencedException is thrown
6. Kafka Streams catches the exception, aborts the transaction (output records rolled back, consumer offset not advanced)
7. Input event is re-delivered to process()
8. stripe.charges.create() is called again → ch_B created (or ch_A returned if idempotency key set and within 24h)

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import com.stripe.Stripe;
import com.stripe.model.Charge;
import com.stripe.param.ChargeCreateParams;
import java.util.Properties;
import static org.apache.kafka.streams.StreamsConfig.EXACTLY_ONCE_V2;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "billing-streams-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
// exactly_once_v2 is the default in Kafka Streams 3.0+.
// It guarantees that output Kafka records and consumer offset commits
// are atomic. It does NOT guarantee that stripe.charges.create()
// executes exactly once — that call is outside the Kafka transaction.
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, EXACTLY_ONCE_V2);

StreamsBuilder builder = new StreamsBuilder();
KStream<String, BillingEvent> billingStream = builder.stream("billing-events");

billingStream.mapValues(event -> {
    // This runs inside the Kafka Streams commit cycle, but OUTSIDE
    // the Kafka producer transaction. The transaction boundaries are:
    //   beginTransaction()
    //   → [process() runs — stripe.charges.create() here is not covered]
    //   → producer.send(outputRecord)          ← inside transaction
    //   → sendOffsetsToTransaction(offsets)    ← inside transaction
    //   → commitTransaction()                  ← can fail here
    //
    // If commitTransaction() fails (ProducerFencedException, broker epoch bump,
    // transaction coordinator failover), Kafka rolls back steps 3-4.
    // The input event is re-delivered. stripe.charges.create() is called again.
    // Without a stable idempotency key, ch_B is created.
    ChargeCreateParams params = ChargeCreateParams.builder()
        .setAmount((long) event.getAmountCents())
        .setCurrency("usd")
        .setCustomer(event.getStripeCustomerId())
        .setDescription("Usage billing for " + event.getBillingPeriod())
        // no idempotency key — exactly_once_v2 does not protect this call
        .build();
    try {
        Charge charge = Charge.create(params);
        return new BillingResult(charge.getId(), "charged");
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}).to("billing-results");

The exactly_once_v2 failure is the most common misconception in the Kafka Streams + Stripe billing space. Teams configure processing.guarantee=exactly_once_v2, verify that the output Kafka topic receives each billing result exactly once, and conclude that their Stripe integration is safe. The Kafka output deduplication is real — downstream consumers reading the billing results topic with isolation.level=read_committed see each result exactly once. But the Stripe HTTP call that produced the result may have been executed multiple times across transaction retries. A cluster that experiences frequent broker leadership elections or a transaction coordinator under memory pressure will trigger this failure mode on every affected commit cycle.

Unlike the SQS, Azure Service Bus, and Azure Event Hubs failure modes — where re-delivery is clearly a consumer-side event — the exactly_once_v2 Kafka transaction failure is invisible at the application level. Kafka Streams handles the ProducerFencedException internally and restarts the commit cycle without surfacing an error to the application code. The process() method runs again with the re-delivered event and no indication that the previous invocation called stripe.charges.create() successfully.

Failure mode 3: DSL groupBy() and selectKey() operations create internal repartition topics — a crash after writing to the repartition topic but before committing the source partition offset injects a duplicate billing event that the downstream stage processes independently

Kafka Streams DSL operations that change the record key — groupBy(), selectKey(), and certain join operations — require that records with the same new key be routed to the same downstream partition. Kafka Streams handles this automatically by writing records to an internal repartition topic named <application-id>-<operation>-repartition (e.g., billing-app-KSTREAM-GROUPBY-repartition). The upstream stage writes records to this internal topic. A downstream consumer group, managed by the same Kafka Streams instance or a separate one, reads from the repartition topic and performs the stateful operation (aggregation, join, count).

The internal repartition topic is a real Kafka topic with the same delivery guarantees as any other Kafka topic. With processing.guarantee=at_least_once, the upstream stage writes to the repartition topic and commits its source consumer offset in two non-atomic steps — the same split that causes failure mode 1 at the state store level, but now applied between two Kafka topics. A crash after the repartition write but before the source offset commit causes the source event to be re-delivered to the upstream stage. The upstream stage re-processes the source event and writes a second copy of the repartition record.

The downstream stage — reading from the repartition topic — now has two copies of the billing event in its partition. Both copies have the same key (the rekeying operation is deterministic) but different offsets and timestamps in the repartition topic. The downstream stage processes each copy as a separate input record and calls stripe.charges.create() for each. If the downstream stage derives its Stripe idempotency key from the repartition topic’s offset or Kafka message metadata — rather than from the billing event’s stable business fields — it produces different idempotency keys for the two copies and creates ch_A and ch_B.

import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import java.time.Duration;

StreamsBuilder builder = new StreamsBuilder();

// Source stream: billing events keyed by arbitrary event ID
KStream<String, UsageEvent> usageStream = builder.stream("usage-events");

// selectKey() repartitions by customer ID. Internally, Kafka Streams
// creates the topic: billing-app-KSTREAM-KEY-SELECT-repartition
// The upstream stage writes records to this internal topic, then
// commits the source ("usage-events") consumer offset.
// With at_least_once, these two operations are non-atomic.
KStream<String, UsageEvent> byCustomer = usageStream
    .selectKey((eventId, event) -> event.getCustomerId());
    // This creates: billing-app-KSTREAM-KEY-SELECT-repartition (internal)
    // Crash after writing to repartition topic, before source offset commit:
    // → source event re-delivered → second copy written to repartition topic.
    // Downstream stage sees two copies → two stripe.charges.create() calls.

// The downstream stage reads from the repartition topic and aggregates usage.
// A billing Processor attached here calls stripe.charges.create() per window.
KGroupedStream<String, UsageEvent> grouped = byCustomer.groupByKey();
// ... aggregate, window, and bill

// UNSAFE downstream billing processor reading from repartition topic:
byCustomer.process(() -> new Processor<String, UsageEvent>() {
    @Override
    public void process(Record<String, UsageEvent> record) {
        String customerId = record.key();
        UsageEvent event = record.value();

        // UNSAFE: idempotency key derived from repartition topic offset.
        // First copy: repartition-topic offset 8821 → key K1
        // Second copy (from upstream crash + re-delivery): offset 8822 → key K2
        // Stripe receives K1 and K2 as separate billing requests → ch_A and ch_B.
        long repartitionOffset = record.offset(); // changes between duplicate copies
        String unsafeKey = customerId + ":" + event.getBillingPeriod()
            + ":" + repartitionOffset; // different for each copy

        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount((long) event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getStripeCustomerId())
            .setDescription("Usage billing for " + event.getBillingPeriod())
            .build();
        try {
            Charge charge = Charge.create(params,
                com.stripe.net.RequestOptions.builder()
                    .setIdempotencyKey(unsafeKey) // different per repartition copy
                    .build());
        } catch (Exception e) { throw new RuntimeException(e); }
    }
});

The repartition failure mode is particularly hard to detect because it operates entirely within Kafka Streams’ internal topic infrastructure. The internal repartition topic name contains the application ID and operation name but is not exposed as a user-configured topic. Kafka consumer lag monitoring on the user-visible source topic (usage-events) does not show the repartition topic’s lag. The duplicate copy in the repartition topic is indistinguishable from a legitimate second billing event at the Kafka topic level — it has a different offset and key-hash-identical routing to the same downstream partition. Only a stable, business-field-derived Stripe idempotency key — and a pre-flight database check — can close this failure mode at the application layer.

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

All three failure modes share the same root cause: stripe.charges.create() is called more than once for the same (customer_id, billing_period) pair, and neither the Kafka Streams commit cycle, the exactly_once_v2 transaction, nor the internal repartition topic infrastructure prevents the second call from creating a new charge when the idempotency key is absent, derived from Kafka metadata that changes across task migrations, or derived from repartition topic offsets that differ between duplicate copies. The fix has three independent components that together close all three failure modes without changing the Kafka Streams topology structure, processing guarantee setting, or partition count.

Layer 1: content-hash idempotency key stable across task migrations and repartition topic duplicates

The Stripe idempotency key must be derived from the billing event’s stable business fields — fields in the event payload that identify the same billing intent regardless of which Kafka Streams task processes it, which partition it occupies in the repartition topic, or how many times it is re-delivered. Fields to exclude from the idempotency key: Kafka partition number (changes when partition count changes or tasks are redistributed), Kafka offset (different for each copy in the repartition topic and changes between consumer group offset resets), Kafka message timestamp (assigned at the time the record was produced to the topic — the repartition topic write has a different timestamp than the source topic write), and Kafka Streams task ID (changes on task migration). All of these are Kafka infrastructure identifiers that track where a record sits in the log, not what billing action it represents.

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.HexFormat;

public class BillingIdempotencyKey {

    public static String compute(String customerId, String billingPeriod) {
        // Derived from stable business identity only.
        // Must NOT include:
        //   - Kafka partition (changes across task migrations and repartitions)
        //   - Kafka offset (different for each repartition topic duplicate copy)
        //   - Kafka message timestamp (set at each topic write — different for
        //     the source write and the repartition write from the re-delivered source)
        //   - Kafka Streams task ID (changes when tasks are redistributed
        //     during scaling events or rolling deployments)
        //   - exactly_once_v2 transaction ID (changes per commit cycle)
        // All of these change across task migrations, repartition copies, and
        // exactly_once_v2 transaction retries.
        String raw = customerId + ":" + billingPeriod + ":kafka-streams-billing";
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(raw.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash).substring(0, 32);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}

Layer 2: vault key with per-billing-period spend cap

The vault key is a per-billing-period Stripe key issued through a spend-cap proxy before the Kafka Streams topology begins processing billing events for that period. The vault key enforces a maximum spend set at 110% of the expected total for the period. For the exactly_once_v2 transaction retry failure mode — where a Stripe call succeeds, the transaction fails, and the event is re-delivered — the content-hash idempotency key deduplicates the retry within Stripe’s 24-hour window. The vault key provides a hard spend cap backstop for the edge case where the consumer group falls behind by more than 24 hours and a billing event processed in a previous run is re-processed after the idempotency window has closed.

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Instant;
import java.time.temporal.ChronoUnit;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.Map;

public class VaultKeyClient {

    private static final HttpClient HTTP = HttpClient.newHttpClient();
    private static final ObjectMapper JSON = new ObjectMapper();
    private static final String KEYBRAKE_API = "https://proxy.keybrake.com";

    public static String issueForBillingPeriod(
            String billingPeriod,
            long expectedTotalCents) throws Exception {

        String body = JSON.writeValueAsString(Map.of(
            "vendor", "stripe",
            "label", "kafka-streams-billing-run-" + billingPeriod,
            "daily_usd_cap", Math.round(expectedTotalCents * 1.10) / 100.0,
            "allowed_endpoints", new String[]{"/v1/charges", "/v1/payment_intents"},
            "expires_at", Instant.now().plus(8, ChronoUnit.HOURS).toString()
        ));

        HttpRequest req = HttpRequest.newBuilder()
            .uri(URI.create(KEYBRAKE_API + "/vault/keys"))
            .header("Authorization", "Bearer " + System.getenv("KEYBRAKE_API_KEY"))
            .header("Content-Type", "application/json")
            .POST(HttpRequest.BodyPublishers.ofString(body))
            .build();

        HttpResponse<String> resp = HTTP.send(req, HttpResponse.BodyHandlers.ofString());
        if (resp.statusCode() != 201) {
            throw new RuntimeException("Vault key issuance failed: " + resp.body());
        }
        return JSON.readTree(resp.body()).get("vault_key").asText();
    }
}

Safe Kafka Streams billing processor with pre-flight database check and write-before-Stripe

import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.ProcessorContext;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.state.KeyValueStore;
import com.stripe.Stripe;
import com.stripe.model.Charge;
import com.stripe.param.ChargeCreateParams;
import com.stripe.net.RequestOptions;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class SafeBillingProcessor implements Processor<String, BillingEvent, String, BillingResult> {

    private ProcessorContext<String, BillingResult> context;
    private KeyValueStore<String, String> billingStatusStore;
    private Connection pgConn;

    @Override
    public void init(ProcessorContext<String, BillingResult> context) {
        this.context = context;
        this.billingStatusStore = context.getStateStore("billing-status-store");
        try {
            this.pgConn = DriverManager.getConnection(
                System.getenv("DATABASE_URL")
            );
        } catch (Exception e) {
            throw new RuntimeException("PostgreSQL connection failed", e);
        }
    }

    @Override
    public void process(Record<String, BillingEvent> record) {
        String customerId = record.key();
        BillingEvent event = record.value();
        String storeKey = customerId + ":" + event.getBillingPeriod();

        String idempotencyKey = BillingIdempotencyKey.compute(
            customerId, event.getBillingPeriod()
        );

        // Pre-flight check: authoritative source is PostgreSQL, not the
        // Kafka Streams state store. The state store is rebuilt from its
        // changelog topic on task migration, which may or may not have
        // the latest write depending on whether the changelog flush completed
        // before the OOM kill. PostgreSQL (WAL-fsynced on commit) survives
        // Kafka Streams task migrations, consumer group rebalances, repartition
        // topic duplicate deliveries, and exactly_once_v2 transaction retries.
        // This closes all three failure modes:
        // (1) at_least_once non-atomic commit: if billing_records was written
        //     before the OOM kill, the restored task's pre-flight check finds
        //     the row and skips stripe.charges.create().
        // (2) exactly_once_v2 transaction retry: if billing_records was written
        //     before the transaction commit failure, the re-delivery's pre-flight
        //     check finds the row and skips stripe.charges.create().
        // (3) Repartition topic duplicate: if the first copy was processed and
        //     billing_records was written, the second copy's pre-flight check
        //     finds the row and skips — regardless of its different repartition offset.
        try (PreparedStatement stmt = pgConn.prepareStatement(
                "SELECT charge_id, status FROM billing_records " +
                "WHERE customer_id = ? AND billing_period = ?")) {
            stmt.setString(1, customerId);
            stmt.setString(2, event.getBillingPeriod());
            try (ResultSet rs = stmt.executeQuery()) {
                if (rs.next()) {
                    // Already billed in PostgreSQL — skip Stripe call entirely.
                    // Also update state store to short-circuit future in-process lookups.
                    billingStatusStore.put(storeKey, rs.getString("charge_id"));
                    context.forward(record.withValue(
                        new BillingResult(customerId, rs.getString("charge_id"), "already_billed")
                    ));
                    return;
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("Pre-flight DB check failed", e);
        }

        String vaultKey = System.getenv("KEYBRAKE_VAULT_KEY");

        Charge charge;
        try {
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount((long) event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Usage billing for " + event.getBillingPeriod())
                .build();
            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .setApiKey(vaultKey)
                .build();
            charge = Charge.create(params, options);
        } catch (com.stripe.exception.CardException e) {
            // Permanent failure — write to billing_records with status='card_declined'.
            // Do NOT leave unprocessed: Kafka Streams will re-deliver indefinitely,
            // creating repeated Stripe card-decline attempts visible to the customer.
            writeToDb(customerId, event.getBillingPeriod(), null,
                      event.getAmountCents(), "card_declined");
            billingStatusStore.put(storeKey, "card_declined");
            context.forward(record.withValue(
                new BillingResult(customerId, null, "card_declined")
            ));
            return;
        } catch (com.stripe.exception.ApiConnectionException e) {
            // Network timeout — ch_A may already exist on Stripe's servers.
            // Do NOT update billing_records or state store.
            // The content-hash idempotency key will return ch_A on re-delivery.
            // Do NOT call context.forward() — do not commit this record's offset
            // as successfully processed. Kafka Streams will re-deliver.
            throw new RuntimeException("Stripe timeout — re-delivery will retry with idempotency key", e);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        // Write to PostgreSQL BEFORE updating the state store and forwarding.
        // Write-before-state-store: if the OOM kill fires between here and
        // billingStatusStore.put(), the pre-flight check on re-delivery finds
        // the billing_records row and skips stripe.charges.create().
        // ON CONFLICT DO NOTHING closes the simultaneous-processor race from
        // repartition topic duplicates processed by concurrent tasks.
        writeToDb(customerId, event.getBillingPeriod(), charge.getId(),
                  event.getAmountCents(), "charged");

        // State store update after PostgreSQL write. If the commit cycle
        // flushes this changelog record before the consumer offset commits,
        // restored tasks can use this as a fast-path check before hitting PostgreSQL.
        billingStatusStore.put(storeKey, charge.getId());

        context.forward(record.withValue(
            new BillingResult(customerId, charge.getId(), "charged")
        ));
    }

    private void writeToDb(String customerId, String billingPeriod,
                           String chargeId, int amountCents, String status) {
        try (PreparedStatement stmt = pgConn.prepareStatement(
                "INSERT INTO billing_records " +
                "  (customer_id, billing_period, charge_id, amount_cents, status, created_at) " +
                "VALUES (?, ?, ?, ?, ?, NOW()) " +
                "ON CONFLICT (customer_id, billing_period) DO NOTHING")) {
            stmt.setString(1, customerId);
            stmt.setString(2, billingPeriod);
            stmt.setString(3, chargeId);
            stmt.setInt(4, amountCents);
            stmt.setString(5, status);
            stmt.executeUpdate();
            pgConn.commit();
        } catch (Exception e) {
            throw new RuntimeException("billing_records write failed", e);
        }
    }

    @Override
    public void close() {
        try { if (pgConn != null) pgConn.close(); } catch (Exception ignored) {}
    }
}

Comparison: protection levels across all three failure modes

Gap analysis

Kafka Streams Punctuator on PunctuationType.WALL_CLOCK_TIME fires independently on each task; after rolling deployment, old-instance Punctuator and new-instance Punctuator may both fire for the same billing window

A Punctuator registered with PunctuationType.WALL_CLOCK_TIME fires based on the wall clock of the Streams instance that holds the task. The Punctuator schedule is not persisted in the state store or the changelog topic — it is reset to zero when a task is migrated to a new instance. If a rolling deployment starts a new instance while the old instance’s Punctuator is in the middle of a billing run, and both instances hold tasks for the same partition briefly during the rebalance window (which can happen when the rebalance protocol allows partial overlap), two Punctuator firings can execute for the same billing window. The content-hash idempotency key derived from customerId + billingPeriod deduplicates the Stripe call within 24 hours, but the PostgreSQL pre-flight check is the primary guard: a UNIQUE (customer_id, billing_period) constraint on billing_records with ON CONFLICT DO NOTHING ensures that only one INSERT succeeds even if two Punctuator firings run concurrently for the same period.

KStream.to() writing to an output topic that is also the source for a separate billing consumer creates two independent billing execution paths, each with its own retry scope

When a Kafka Streams topology writes billing results to a Kafka output topic via KStream.to() and a separate consumer (a different Kafka Streams application or a plain Kafka consumer) reads from that output topic to trigger downstream billing operations, the two systems have independent retry scopes. A failure in the downstream consumer does not roll back the upstream Kafka Streams write. The content-hash idempotency key must be the same in both the upstream Streams processor and the downstream billing consumer: if the upstream processor uses customer_id + billing_period + kafka-streams-billing as the idempotency key component and the downstream consumer uses a different key derivation, a retry in the downstream consumer will create a charge with a key that Stripe has not seen before (even if the upstream key already deduplicates the original charge). Align idempotency key derivation across all billing stages that call Stripe for the same billing intent.

Kafka Streams exactly_once_v2 with multiple Streams instances: transaction coordinator rebalance during scaling can cause two instances to hold the same task briefly, both calling stripe.charges.create() before the old instance releases its state store

During a scaling event (adding a new Streams instance to increase parallelism), the Kafka Streams rebalance protocol redistributes tasks. The old instance may hold a task and call stripe.charges.create() in the closing milliseconds before it releases the task, while the new instance has already acquired the task assignment and begins processing from the same partition. Both instances hold a valid transactional producer for a brief overlap window. The exactly_once_v2 fencing mechanism ensures only one transaction commits (the old instance’s epoch is bumped when the new instance initializes its transactional producer), but the stripe.charges.create() call from the old instance may have already completed. The content-hash idempotency key handles this: both instances use the same key for the same billing event, and Stripe returns ch_A for both.

Kafka Streams state store used as a dedup guard has task-local scope: a billing event rekeyed via selectKey() to a partition served by a different task has no visibility into the source task’s state store

Each Kafka Streams state store is local to the task that owns it. State stores are partitioned: task T0 holds the RocksDB store for partition 0, task T1 for partition 1, and so on. A billing event that is rekeyed via selectKey(customer_id) may be routed to a partition served by a different task than the source event’s task. If both the source task and the downstream task maintain their own billing-status state stores, a dedup check in the downstream task’s store has no record of billing events that were processed by the source task (which has its own store keyed differently). PostgreSQL — shared across all task instances and all Streams applications — is the only correct dedup store for billing events processed across multiple tasks and repartition operations.

FAQ

Does setting processing.guarantee=exactly_once_v2 mean I can remove the Stripe idempotency key from my billing processor?

No. The exactly_once_v2 processing guarantee ensures that output Kafka records are written to downstream topics exactly once (atomically with the consumer offset commit, using Kafka producer transactions). It does not guarantee that the code inside your process() method executes exactly once. stripe.charges.create() is an external HTTP call that runs before the Kafka transaction commits. If the transaction commit fails — due to broker epoch bumps, transaction coordinator failover, or network partition between the Streams instance and the coordinator — the input event is re-delivered and process() runs again. The Stripe idempotency key is required regardless of the Kafka Streams processing guarantee setting. The guarantee setting controls Kafka-to-Kafka write semantics; your billing code is responsible for the Stripe-specific idempotency.

Can I use Kafka Streams’ StateStore as the primary dedup guard instead of PostgreSQL?

A KeyValueStore backed by a RocksDB changelog topic can serve as a fast-path dedup check inside the process() method, but it is not a sufficient primary guard for three reasons. First, the state store write and the Stripe call are ordered sequentially in your code, and an OOM kill between them leaves the state store without the billing record — the restored task’s state store check misses it and re-charges the customer. Second, state store entries are task-local: a billing event that arrives via a repartition topic copy processed by a different task (with a separate state store) has no visibility into the original event’s state store entry. Third, state store changelog restoration is async on task startup — there is a brief window before restoration completes where the state store appears empty and a billing processor that skips the pre-flight PostgreSQL check could proceed to charge. PostgreSQL (with a UNIQUE constraint and ON CONFLICT DO NOTHING) is the authoritative dedup store because it is shared across all tasks and persists independently of Kafka Streams task lifecycle.

Does Kafka Streams’ standby.replicas setting prevent billing duplicates on task migration?

num.standby.replicas tells Kafka Streams to maintain shadow copies of state stores on standby task instances, reducing the time to restore state after task migration (the standby instance is already up-to-date with the changelog topic and can take over quickly). Standby replicas reduce the duration of the re-delivery window — a faster state restore means fewer events need to be re-processed from the last committed consumer offset. But standby replicas do not eliminate re-delivery: with at_least_once processing, the non-atomic commit cycle creates a window where state store changelog records are committed to Kafka but consumer offsets are not. Standby replicas help the state store recover, but billing events processed in that window are still re-delivered to the new task owner. The idempotency key and pre-flight database check are required regardless of the standby replica setting.

How should I issue the vault key for a Kafka Streams billing run that processes events continuously vs. in discrete billing periods?

For billing systems that process usage events continuously (e.g., a Kafka Streams topology that charges customers in real time as each billable event arrives), issue a vault key per customer per billing period and store the vault key in the Kafka Streams state store keyed by customer_id + billing_period. Set the vault key spend cap to the expected charge amount for that period (plus 10% buffer) and TTL to the period duration plus the expected consumer group lag window. For systems that process billing in discrete batch windows (e.g., a Punctuator that fires at midnight to charge all customers for the day), issue a single vault key per billing period before the Punctuator fires, scoped to the aggregate expected total for all customers in that period. Pass the vault key through a Kafka Streams header or a state store entry that the billing Processor reads before calling Stripe. Do not use a single long-lived vault key shared across all billing periods — the spend cap would need to cover all future periods, making the cap meaningless as a governance control.

Do Kafka Streams topology changes (adding a new operator, changing a state store name) affect the idempotency key or cause duplicate billing?

Topology changes that modify internal topic names (changing the application ID, renaming a state store, adding or removing operations that create repartition topics) cause Kafka Streams to generate new internal topic names. Existing events in the old internal topics are not automatically migrated to new topic names. If a topology upgrade changes a repartition topic name, the new topology starts reading from the new (empty) repartition topic while the old repartition topic still contains unprocessed billing events. Those events are not re-billed — they are simply not processed by the new topology (which reads from a different topic). The idempotency key is not affected by topology changes because it is derived from business fields in the event body, not from topic names or task IDs. The concern is instead that a topology change may leave unbilled customers whose events were in the old repartition topic. Use Kafka Streams’ topology compatibility mode and verify that all events in active repartition topics are fully processed before deploying a topology change that renames internal topics.

Can I use Kafka Streams’ KafkaStreams.cleanUp() to reset state before re-running a billing job?

KafkaStreams.cleanUp() deletes local RocksDB state store directories and resets the Kafka Streams application to rebuild state from scratch. It does not reset consumer group offsets — the input partitions resume from their last committed offsets. Running cleanUp() before a re-run causes the billing processor to re-process events from the last committed offset with an empty local state store (no fast-path dedup guard from the RocksDB check). The pre-flight PostgreSQL check remains the authoritative dedup guard in this scenario: even with an empty state store, the billing_records table correctly identifies already-billed customers and prevents duplicate charges. Never use cleanUp() on a production billing Kafka Streams application without verifying that the pre-flight database check is in place — a clean state store with uncommitted consumer offsets is the same initial state as a fresh deployment against a topic with historical billing events.