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

Apache Kafka is a distributed event-streaming platform built around at-least-once delivery guarantees. Teams wire Kafka consumer groups into Stripe billing pipelines to process usage events, subscription renewals, and metered-billing triggers at scale: a consumer reads from a billing-events topic, calls stripe.charges.create() for each event, writes the charge result to a database, and commits the Kafka offset. Three Kafka-specific behaviors introduce Stripe double-charge failure modes that consumer lag dashboards, offset monitoring, and dead letter queues do not surface as billing risk.

This post covers those three failure modes with kafka-python code, content-hash idempotency keys, per-billing-period vault keys via a spend-cap proxy, and manual offset commit after the database write — the two-layer governance pattern that eliminates all three without replacing Kafka’s delivery model.

Failure mode 1: at-least-once delivery causes duplicate charge on consumer crash before offset commit

Kafka’s delivery guarantee is at-least-once: a message is guaranteed to be delivered to a consumer, but may be delivered more than once. The mechanism that makes this possible is the consumer’s committed offset — the position in the topic partition up to which the consumer has successfully processed messages. If a consumer crashes, the next consumer assigned to that partition starts reading from the last committed offset, which may be before some messages that the crashed consumer had already processed.

With enable_auto_commit=True and auto_commit_interval_ms=5000 (the kafka-python defaults), offsets are committed to the Kafka broker every five seconds in the background. A consumer that processes billing events sequentially — calling stripe.charges.create() for each event and then writing the charge ID to a database — can crash at any point between auto-commit intervals. If it crashes after the Stripe call for event N succeeded but before the next auto-commit fires, the offset for event N is not recorded. On restart (or when a new consumer in the same group picks up the partition), that consumer starts from the last committed offset — which is before event N — and re-processes it. A second call to stripe.charges.create() with no idempotency key creates ch_B for the same customer and billing period.

# UNSAFE: Kafka billing consumer with auto-commit and no idempotency key
# A crash between stripe.charges.create() and the next auto-commit fires
# causes the same billing event to be reprocessed on consumer restart

from kafka import KafkaConsumer
import json, os, stripe

consumer = KafkaConsumer(
    'billing-events',
    bootstrap_servers=['kafka:9092'],
    group_id='billing-consumer',
    auto_offset_reset='earliest',
    enable_auto_commit=True,          # offsets committed every 5s in background
    auto_commit_interval_ms=5000,
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
)

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # UNSAFE: unrestricted, shared

for msg in consumer:
    event = msg.value

    # Stripe call succeeds -- ch_A created
    charge = stripe.Charge.create(
        amount=event["amount_cents"],
        currency="usd",
        customer=event["stripe_customer_id"],
        description=f"Billing for {event['billing_period']}",
        # no idempotency_key -- Stripe generates a new request ID each call
    )

    # Consumer crashes here (OOM, pod eviction, SIGKILL) before the
    # 5-second auto-commit interval fires. On restart, the consumer reads
    # from the last committed offset, re-processes this event, and creates ch_B.
    write_charge_to_db(
        customer_id=event["customer_id"],
        charge_id=charge["id"],
        billing_period=event["billing_period"],
    )
    # auto-commit fires asynchronously -- not guaranteed to run before crash

Kafka’s consumer lag metric shows zero lag once the next consumer catches up — the duplicate charge is invisible to Kafka monitoring because both the first and second consumer successfully processed the event. The topic offset for the billing event advances normally in both cases. Stripe’s dashboard shows two charges for the same customer in the same billing period, each with a different request_id and a different creation timestamp. The second charge has no corresponding database row in the billing records table (the first consumer crashed before writing it), so reconciliation queries that look for “charges without a matching DB record” will find ch_B but not ch_A.

Failure mode 2: max_poll_interval_ms exceeded during sequential Stripe calls — consumer evicted mid-batch, in-progress events reprocessed

Kafka’s consumer group protocol uses a liveness timer called max.poll.interval.ms. The broker expects the consumer to call poll() at least once within this interval. If the consumer fails to poll within the window — because it is spending the entire interval processing a previous batch of messages — the broker treats the consumer as dead, evicts it from the group, and triggers a partition rebalance. The evicted consumer’s partition is reassigned to another active consumer, which starts reading from the last committed offset for that partition.

In a billing consumer that processes messages sequentially, each Stripe API call takes between 200ms and 2 seconds (accounting for network latency, Stripe’s server processing, and occasional retries on connection errors). With max_poll_records=500 and each call taking 500ms, a full batch takes 250 seconds to process. If max_poll_interval_ms=60000 (60 seconds, a common value for monitoring responsiveness), the consumer exceeds the limit 60 seconds into processing its 500-message batch — after approximately 120 events. Kafka evicts the consumer and rebalances the partition. The last committed offset may have been set 60+ seconds ago, meaning the reassigned consumer re-reads and re-processes all 500 events in the batch, including the 120 that were already charged.

# UNSAFE: Kafka billing consumer with large batch size and slow Stripe calls
# max_poll_records=500 with 500ms per Stripe call = 250 seconds per batch
# max_poll_interval_ms=60000 -- consumer is evicted 60 seconds in,
# after ~120 events are processed. All 500 are reprocessed by the new consumer.

from kafka import KafkaConsumer
import json, os, stripe, time

consumer = KafkaConsumer(
    'billing-events',
    bootstrap_servers=['kafka:9092'],
    group_id='billing-consumer',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    auto_commit_interval_ms=5000,
    max_poll_interval_ms=60000,   # DANGER: too short for sequential Stripe calls
    max_poll_records=500,         # DANGER: 500 events × 500ms = 250s per batch
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
)

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # UNSAFE: unrestricted

for msg in consumer:
    event = msg.value

    # Each Stripe call takes 200-2000ms
    charge = stripe.Charge.create(
        amount=event["amount_cents"],
        currency="usd",
        customer=event["stripe_customer_id"],
        description=f"Billing for {event['billing_period']}",
        # no idempotency_key
    )
    write_charge_to_db(event["customer_id"], charge["id"], event["billing_period"])
    # After 60 seconds, Kafka evicts this consumer. The new consumer starts
    # from the last committed offset and re-processes the entire batch.

The consumer group coordinator sees a clean rebalance: the evicted consumer sends a LeaveGroup request (or the heartbeat times out), the coordinator reassigns the partition, and the new consumer joins normally. Consumer lag briefly spikes during the rebalance, then drops back to zero as the new consumer catches up. No error is raised. The billing pipeline logs show a gap in processing times, but no exception. Stripe’s dashboard shows a cluster of duplicate charges created within seconds of each other — one set from the evicted consumer, one set from the replacement — separated by the rebalance latency.

Failure mode 3: consumer group offset reset replays all billing events in the retention window

Kafka topics retain messages for a configurable retention period, commonly 7 days for operational topics and up to 30 days for audit or billing event topics. A consumer group’s committed offsets are stored in the internal __consumer_offsets topic with a shorter retention window (default 7 days of inactivity). If the consumer group stops consuming for longer than the offset retention period, the offsets are deleted. On restart, the consumer falls back to the auto_offset_reset policy — either earliest (replay from the start of the retention window) or latest (skip all unprocessed messages).

Teams also deliberately reset consumer group offsets during incident response or deploy rollouts: kafka-consumer-groups.sh --group billing-consumer --reset-offsets --to-earliest --execute --topic billing-events. An operations engineer who runs this command to “reprocess billing events after a code bug” or “replay from the start to verify the new consumer logic” without first disabling the Stripe calls causes every customer in the billing events topic to be charged a second time. With a 30-day retention window, that is every customer billed in the last month.

# DANGEROUS: offset reset replays all billing events in the retention window
# Run by an ops engineer to reprocess events after a bug fix:
#
#   kafka-consumer-groups.sh \
#     --bootstrap-server kafka:9092 \
#     --group billing-consumer \
#     --reset-offsets --to-earliest --execute \
#     --topic billing-events
#
# Result: the consumer group starts from offset 0 on each partition.
# Every billing event in the 30-day retention window is reprocessed.
# Every customer in that window is charged a second time.
# No error is raised. Consumer lag is zero after the replay completes.

# A new deployment that accidentally changes the group.id has the same effect:
consumer = KafkaConsumer(
    'billing-events',
    bootstrap_servers=['kafka:9092'],
    group_id='billing-consumer-v2',  # NEW group.id -- no committed offsets exist
    auto_offset_reset='earliest',     # falls back to start of retention window
    ...
)

The safe approach for offset resets is to add a “dry-run” flag to the billing consumer that processes events and logs what it would charge, without calling Stripe, to verify the code logic. Only after the dry-run output is audited should the flag be removed and the consumer run against Stripe. With idempotency keys in place, an accidental full replay is still safe: Stripe returns the original charge for every event within the 24-hour idempotency window. Events outside that window create new charges — which is why the content-hash idempotency key combined with a pre-flight database check is the complete solution.

# SAFE: Kafka billing consumer with content-hash idempotency key,
# manual offset commit, vault key spend cap, and pre-flight DB check

from kafka import KafkaConsumer, TopicPartition
import hashlib, json, os, httpx, stripe

VAULT_KEY   = os.environ["KEYBRAKE_VAULT_KEY"]  # per-billing-period vault key
PROXY_BASE  = "https://proxy.keybrake.com/stripe"

consumer = KafkaConsumer(
    'billing-events',
    bootstrap_servers=['kafka:9092'],
    group_id='billing-consumer',
    auto_offset_reset='earliest',
    enable_auto_commit=False,         # SAFE: manual commit after DB write
    max_poll_interval_ms=600000,      # 10 minutes -- generous for Stripe calls
    max_poll_records=50,              # small batches to stay within poll interval
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
)

stripe.api_key  = VAULT_KEY
stripe.api_base = PROXY_BASE

for msg in consumer:
    event = msg.value

    # Content-hash idempotency key: stable across consumer restarts and
    # partition rebalances. Does NOT include Kafka offset, partition, or
    # consumer group -- those change on rebalance.
    raw_key = (
        f"{event['customer_id']}:{event['amount_cents']}"
        f":{event['billing_period']}:kafka-billing"
    )
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    # Pre-flight check: skip Stripe call if this period was already billed.
    # Protects against offset resets outside the 24h idempotency window.
    if already_billed(event["customer_id"], event["billing_period"]):
        consumer.commit()  # advance offset so we don't re-read this event
        continue

    try:
        charge = stripe.Charge.create(
            amount=event["amount_cents"],
            currency="usd",
            customer=event["stripe_customer_id"],
            description=f"Billing for {event['billing_period']}",
            idempotency_key=idempotency_key,  # SAFE: same key on every replay
        )
    except stripe.error.CardError:
        # Permanent failure -- write declined result, advance offset
        write_declined_to_db(event["customer_id"], event["billing_period"])
        consumer.commit()
        continue
    except stripe.error.InvalidRequestError as e:
        write_error_to_db(event["customer_id"], event["billing_period"], str(e))
        consumer.commit()
        continue

    # Write BEFORE commit: if write succeeds but commit fails, the pre-flight
    # check on the next re-read will find the existing record and skip Stripe.
    write_charge_to_db(
        customer_id=event["customer_id"],
        charge_id=charge["id"],
        billing_period=event["billing_period"],
    )

    # Commit AFTER successful DB write: at-least-once but idempotent on replay.
    consumer.commit()

The critical ordering: write to the database before committing the offset. If the database write succeeds but the consumer crashes before the consumer.commit() call, the next consumer re-reads the event and the pre-flight check finds the existing database record and skips the Stripe call. The message is re-consumed but no duplicate charge occurs. If you commit the offset before the database write, a crash between the two leaves a committed offset with no database record — the event is silently skipped and the charge never happens.

Issuing the vault key before the consumer loop

The vault key is issued once per billing period before the consumer loop starts — not inside the message processing loop. Query the database for the expected billing total for the period, issue a vault key capped at 110% of that total, and pass it to the consumer as an environment variable or constructor argument. This ensures that a data error in amount_cents that propagates to all events in the period hits the cap before the majority of customers are charged, even though Kafka consumers process messages one at a time rather than in a parallel fan-out.

# Issue vault key before starting the consumer loop

import httpx, os

def issue_billing_vault_key(billing_period: str) -> str:
    expected_total_cents = query_expected_billing_total(billing_period)

    resp = httpx.post(
        "https://api.keybrake.com/vault-keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "label": f"kafka-billing-{billing_period}",
            "vendor": "stripe",
            "spend_cap_usd": round(expected_total_cents / 100 * 1.10, 2),
            "allowed_endpoints": ["POST /v1/charges"],
            "ttl_seconds": 7200,  # 2h: enough for the full billing run + retries
        },
        timeout=10.0,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# In the billing consumer entry point:
billing_period = os.environ["BILLING_PERIOD"]  # e.g. "2026-07"
vault_key = issue_billing_vault_key(billing_period)
os.environ["KEYBRAKE_VAULT_KEY"] = vault_key
# Now start the consumer loop

Gap analysis: four more Kafka–Stripe failure modes to address

The three failure modes above cover the most common Kafka billing incidents. Four additional scenarios are worth addressing before production:

1. Dead letter queue (DLQ) retry without carrying the idempotency key. A common pattern sends failed billing events to a billing-events-dlq topic for later retry. If the DLQ consumer reconstructs a new idempotency key from the replayed event (or omits it entirely), a stripe.error.APIConnectionError from the original consumer that landed in the DLQ may be retried as a fresh Stripe call — even though APIConnectionError does not mean the charge was not created; it means the response was lost in transit. ch_A may have been created before the connection error, and the DLQ retry creates ch_B. The fix: include the idempotency key in the DLQ event payload and carry it through all retry paths.

2. Producer idempotency does not protect consumer-side Stripe calls. Kafka’s enable.idempotence=true producer setting prevents the producer from writing duplicate messages to Kafka due to retried produce requests. Teams sometimes assume this means events are “idempotent” end-to-end. It is not. Kafka producer idempotency is scoped to the producer-to-broker write path. A consumer receiving that same event and calling stripe.charges.create() twice — due to rebalance, crash, or offset reset — creates two charges regardless of the producer’s idempotency setting.

3. Kafka Streams application restart with a new application.id replays all events. Kafka Streams applications use the application.id as the consumer group ID for their internal topics. If the application is redeployed with a changed application.id (for example, during a version migration that appends -v2 to the ID), Kafka Streams starts as a new consumer group with no committed offsets. With auto.offset.reset=earliest, the stream processor replays the entire topic from the beginning of the retention window, calling Stripe for every event in that window. For a billing Kafka Streams application, this means every customer in the retention window is charged again.

4. Vault key TTL must cover the full consumer run including partition rebalance recovery time. If the vault key expires while the consumer group is processing a long billing run (or during a rebalance that pauses processing), the Keybrake proxy returns 401 Unauthorized for subsequent calls. Without explicit handling, the kafka-python client raises an exception and the consumer enters an error state. Catch stripe.error.AuthenticationError explicitly and write it to the database as a permanent error rather than re-raising (which would skip the consumer.commit() and cause the event to be re-read after the TTL error has already been written). Set the TTL to at least twice the expected wall-clock time for the full billing run, accounting for rebalance overhead.

Approach comparison

pytest enforcement suite

import hashlib, pytest
from unittest.mock import patch, MagicMock

# 1. Content-hash idempotency key excludes Kafka-specific fields
def test_idempotency_key_excludes_kafka_fields():
    event = {
        "customer_id": "cus_001",
        "amount_cents": 4900,
        "billing_period": "2026-07",
        "_kafka_offset": 12345,   # must NOT be in the key
        "_kafka_partition": 2,
    }
    raw = f"{event['customer_id']}:{event['amount_cents']}:{event['billing_period']}:kafka-billing"
    key = hashlib.sha256(raw.encode()).hexdigest()[:32]

    # Simulate a rebalance: same event arrives on a different partition/offset
    event_replayed = dict(event, _kafka_offset=99999, _kafka_partition=5)
    raw_replayed = (
        f"{event_replayed['customer_id']}:{event_replayed['amount_cents']}"
        f":{event_replayed['billing_period']}:kafka-billing"
    )
    key_replayed = hashlib.sha256(raw_replayed.encode()).hexdigest()[:32]

    assert key == key_replayed, "Idempotency key must be stable across partition/offset changes"

# 2. Pre-flight check short-circuits Stripe call for already-billed period
def test_preflight_check_skips_stripe_call(mock_db, mock_stripe):
    mock_db.already_billed.return_value = True  # billing period already processed

    process_billing_event(
        event={"customer_id": "cus_001", "amount_cents": 4900, "billing_period": "2026-07",
               "stripe_customer_id": "cus_stripe_001"},
        db=mock_db,
    )

    mock_stripe.Charge.create.assert_not_called()

# 3. Consumer crash simulation: same event reprocessed, Stripe returns original charge
def test_consumer_crash_reprocess_returns_original_charge(mock_db, mock_stripe):
    event = {"customer_id": "cus_002", "amount_cents": 9900,
             "billing_period": "2026-07", "stripe_customer_id": "cus_stripe_002"}
    raw_key = f"{event['customer_id']}:{event['amount_cents']}:{event['billing_period']}:kafka-billing"
    expected_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    mock_db.already_billed.return_value = False
    mock_stripe.Charge.create.return_value = {"id": "ch_original"}

    # First processing
    process_billing_event(event, db=mock_db)
    first_call_key = mock_stripe.Charge.create.call_args[1]["idempotency_key"]
    assert first_call_key == expected_key

    # Simulate crash + restart: same event reprocessed
    mock_stripe.Charge.create.reset_mock()
    mock_db.already_billed.return_value = False  # DB write also failed on first crash
    process_billing_event(event, db=mock_db)
    second_call_key = mock_stripe.Charge.create.call_args[1]["idempotency_key"]

    assert first_call_key == second_call_key  # Stripe deduplicates; only ch_original returned

# 4. Vault key spend cap rejects a batch where amount_cents has a 100x data error
def test_vault_key_spend_cap_blocks_data_error(mock_proxy):
    # Expected: 100 customers × $49.00 = $4,900 total. Vault key capped at $5,390 (110%).
    # Data error: amount_cents = 490000 (i.e., $4,900 per customer instead of $49.00)
    # After ~1 event, the cap is exhausted and the proxy returns 402.
    mock_proxy.post.side_effect = [
        MagicMock(status_code=200, json=lambda: {"id": "ch_001"}),  # first event succeeds
        MagicMock(status_code=402, json=lambda: {"error": "spend_cap_exceeded"}),  # second blocked
    ]

    results = process_billing_batch(
        events=[{"customer_id": f"cus_{i}", "amount_cents": 490000,
                 "billing_period": "2026-07"} for i in range(100)],
        vault_key="vk_test_cap",
    )

    assert results[0]["status"] == "charged"
    assert results[1]["status"] == "cap_exceeded"
    assert sum(1 for r in results if r["status"] == "charged") == 1

# 5. DLQ retry carries original idempotency key
def test_dlq_retry_carries_original_idempotency_key():
    original_event = {
        "customer_id": "cus_003",
        "amount_cents": 1900,
        "billing_period": "2026-07",
        "stripe_customer_id": "cus_stripe_003",
    }
    raw = f"{original_event['customer_id']}:{original_event['amount_cents']}:{original_event['billing_period']}:kafka-billing"
    original_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

    # DLQ event must preserve the idempotency key from the original event
    dlq_event = dict(original_event, _idempotency_key=original_key)

    # DLQ consumer must use the preserved key, not generate a new one
    assert dlq_event["_idempotency_key"] == original_key

Frequently asked questions

Q: Can I use topic:partition:offset as the Stripe idempotency key to avoid computing a hash?
A: Only within Stripe’s 24-hour idempotency window. After 24 hours, the key expires and a consumer group offset reset or a Kafka Streams application restart with a new application.id replays the same message at the same partition:offset — but Stripe treats it as a new request and creates a duplicate charge. The content-hash key derived from business inputs (customer ID, amount, billing period) is durable across replay scenarios regardless of timing.

Q: Does Kafka’s exactly-once semantics (EOS) prevent duplicate Stripe charges?
A: No. Kafka EOS prevents duplicate messages in Kafka-to-Kafka pipelines — specifically, it ensures a producer’s messages are written exactly once to a Kafka topic, and a transactional consumer can consume and produce within a single atomic transaction. Calling stripe.charges.create() is an external side effect outside the Kafka transaction boundary. The Kafka transaction coordinator does not know about Stripe calls, cannot roll them back, and cannot prevent them from being called twice on consumer rebalance or crash.

Q: Should I use enable.auto.commit=false and commit after every message, or batch commits?
A: Commit after every message processed (after both Stripe call and DB write succeed). Batch commits are more efficient but create a larger window where a crash causes reprocessing. With content-hash idempotency keys and a pre-flight DB check, reprocessing is safe, so batch commits are acceptable for performance — but the semantics are easier to reason about when you commit once per message.

Q: How long should the vault key TTL be for a Kafka billing consumer?
A: At least twice the expected wall-clock time for the complete billing run, measured from key issuance to the last expected offset commit. For a consumer processing 10,000 events at 500ms per Stripe call, that is 5,000 seconds (83 minutes). Set the TTL to 3 hours. Include rebalance recovery time: a group with 4 consumers rebalancing can pause processing for 30-90 seconds per rebalance event. Add 30 minutes of headroom for unexpected rebalances during the billing run.

Q: What is the right max_poll_records value for a billing consumer?
A: Size it so that max_poll_records × p99_stripe_latency_ms < max_poll_interval_ms / 2. If your p99 Stripe latency is 1,500ms and your max_poll_interval_ms is 300,000ms (5 minutes), you can safely process up to 100 events per poll batch (100 × 1,500ms = 150,000ms < 150,000ms). Leave the factor-of-two buffer for unexpected Stripe latency spikes. Reduce max_poll_records rather than increasing max_poll_interval_ms — a longer poll interval delays consumer failure detection for the group coordinator.

Q: How do I safely replay Kafka billing events for debugging without creating duplicate charges?
A: Use a separate consumer group with a DRY_RUN=true environment variable. The consumer reads events, computes the idempotency key, checks the pre-flight database, and logs what it would charge — without calling Stripe. Once the dry-run output is audited, switch to the production group. Never reset offsets on the production billing consumer group to “replay” events; use the read-only consumer group instead. With idempotency keys and pre-flight checks in place, the production consumer can be restarted from any offset safely — but a dry-run pass first reduces the audit surface.