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

Apache Flink is a distributed stream processing framework built around stateful operators, event-time semantics, and checkpoint-based fault tolerance. AI teams use Flink to process high-volume event streams in real time: charging customers based on usage events, invoking Stripe from AI agent decision pipelines running as Flink jobs, and running billing aggregations over sliding windows. When a Flink operator calls stripe.charges.create(), Flink's recovery model introduces three specific failure modes that neither Stripe restricted keys nor standard exception handling prevents.

This post covers all three failure modes with Python DataStream API code, and the two-layer governance pattern — content-hash idempotency keys plus per-pipeline vault keys via a spend-cap proxy — that eliminates all three without restructuring your Flink topology.

Failure mode 1: checkpoint recovery replays records through billing operators from the last checkpoint

Flink's fault tolerance is built on distributed snapshots called checkpoints. At configurable intervals, Flink snapshots the state of every operator in the job graph and commits the read offset of every source (Kafka consumer group offset, file position, etc.) to durable storage. When a task fails — a TaskManager OOM, a network partition, a container eviction — Flink restores all operators from their last successful checkpoint state, seeks the source back to the committed offset, and replays all records that arrived after the checkpoint.

For stateful computations — counters, aggregations, window results stored in Flink's managed state backends — this replay produces exactly-once semantics because the state was also snapshotted. But for external API calls made inside operator processElement() methods — like stripe.charges.create() — there is no snapshot. Flink cannot know whether the call completed before the failure. It replays the record unconditionally. Without an idempotency key, Stripe creates a second charge for the same customer, for the same billing event.

# billing_operator.py — UNSAFE: no idempotency key, checkpoint replay creates duplicate charges
from pyflink.datastream import StreamExecutionEnvironment, MapFunction
from pyflink.common.typeinfo import Types
import stripe
import os

class BillingOperator(MapFunction):
    """Charges customers from a stream of billing events."""

    def open(self, runtime_context):
        # UNSAFE: bare Stripe key, no idempotency key strategy defined
        stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

    def map(self, billing_event):
        customer_id = billing_event["customerId"]
        amount_cents = billing_event["amountCents"]
        billing_period = billing_event["billingPeriod"]

        # UNSAFE: no idempotency key.
        # If any downstream task fails after this call — a Kafka sink write timeout,
        # a Flink state backend flush error, a TaskManager eviction — Flink restores
        # from the last checkpoint and replays this record. stripe.charges.create()
        # fires again for the same billing_event → ch_B created.
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Billing {billing_period}",
        )

        return {"customerId": customer_id, "chargeId": charge["id"], "status": "charged"}

The severity depends on checkpoint interval. With a 10-minute checkpoint interval, a task failure at minute 9 replays the last 9 minutes of billing events through the operator. For a high-throughput billing stream — hundreds of customers per minute — that can mean thousands of duplicate charges before the job recovers. Because Flink reports successful recovery (all operators restored, job running), the duplicates are invisible until customers dispute charges or a reconciliation job surfaces the Stripe charge count mismatch.

The fix is a content-hash idempotency key derived deterministically from the billing event's immutable fields — customer ID, amount, and billing period — combined with a Flink-specific salt. Because the same billing event always produces the same idempotency key regardless of which checkpoint replay cycle the operator is in, Stripe returns the original charge object on replay rather than creating a new one.

# billing_operator.py — SAFE: content-hash idempotency key stable across checkpoint replays
from pyflink.datastream import MapFunction
import stripe
import os
import hashlib

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:flink-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

class BillingOperator(MapFunction):
    """Charges customers from a stream of billing events. Safe across checkpoint replays."""

    def open(self, runtime_context):
        # Use per-pipeline vault key (not bare Stripe key) for spend-cap enforcement
        self._vault_key = os.environ["KEYBRAKE_VAULT_KEY"]

    def map(self, billing_event):
        customer_id = billing_event["customerId"]
        amount_cents = billing_event["amountCents"]
        billing_period = billing_event["billingPeriod"]

        # Idempotency key is deterministic from event content — identical on every replay
        idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

        stripe_client = stripe.Stripe(
            api_key=self._vault_key,
            base_url="https://proxy.keybrake.com/stripe",
        )

        try:
            charge = stripe_client.charges.create(
                params={
                    "amount": amount_cents,
                    "currency": "usd",
                    "customer": customer_id,
                    "description": f"Billing {billing_period}",
                },
                options={"idempotency_key": idempotency_key},
            )
            return {
                "customerId": customer_id,
                "chargeId": charge.id,
                "status": "charged",
                "billingPeriod": billing_period,
            }
        except stripe.error.CardError as e:
            # Permanent error: surface as a result record, do not raise
            # Raising here would cause Flink to retry this operator invocation
            return {
                "customerId": customer_id,
                "chargeId": None,
                "status": f"card_declined:{e.code}",
                "billingPeriod": billing_period,
            }

Return permanent Stripe errors as result records rather than raising exceptions. When a MapFunction raises, Flink considers the task as failed and may trigger checkpoint recovery — which would replay the billing event and call the operator again. For permanent failures (card declined, invalid customer ID, currency mismatch), raising is counterproductive: the same record will fail on every replay, producing an infinite retry loop until the Flink job's restart strategy limit is reached. Return the error information as a structured output record so downstream operators or sinks can route it to a dead-letter stream or alerting system.

Failure mode 2: AsyncDataStream.unorderedWait() fan-out shares one unrestricted STRIPE_SECRET_KEY across N concurrent in-flight Stripe calls

Flink's async I/O API (AsyncDataStream.unorderedWait() or orderedWait()) allows operators to make external calls without blocking the main processing thread. Instead of one Stripe call at a time, the async operator maintains a pool of up to capacity concurrent in-flight requests. In a billing pipeline with capacity=100, up to 100 stripe.charges.create() calls are in flight simultaneously — one per billing event from the input stream.

All 100 concurrent calls share the same STRIPE_SECRET_KEY environment variable loaded in open(). The key has no per-request or per-event spending limit. If a unit calculation error in the upstream data pipeline produces an incorrect amountCents field — wrong currency factor, unhandled decimal precision, a missing discount coefficient — every concurrent in-flight call charges the wrong amount simultaneously. Because the async operator maintains up to 100 requests in flight before any results return, the error propagates to all 100 concurrent charges before the first result is processed and any alerting can fire. By the time the first charge result is received and a downstream monitoring operator detects an anomaly, 99 more charges at the wrong amount are already submitted to Stripe.

# async_billing.py — UNSAFE: all concurrent async Stripe calls share one unrestricted key
from pyflink.datastream.functions import AsyncFunction, ResultFuture
from pyflink.datastream import AsyncDataStream
import asyncio
import aiohttp
import os

class AsyncBillingFunction(AsyncFunction):
    """Async Stripe billing operator — processes up to capacity events concurrently."""

    def open(self, runtime_context):
        # UNSAFE: bare Stripe key shared across all concurrent requests.
        # A data error in amount_cents hits all concurrent in-flight calls
        # before any result returns and any cap can fire.
        self._stripe_key = os.environ["STRIPE_SECRET_KEY"]
        self._session = None

    async def async_invoke(self, billing_event, result_future: ResultFuture):
        customer_id = billing_event["customerId"]
        amount_cents = billing_event["amountCents"]
        billing_period = billing_event["billingPeriod"]

        if self._session is None:
            self._session = aiohttp.ClientSession()

        # UNSAFE: no per-event spend cap, no idempotency key
        async with self._session.post(
            "https://api.stripe.com/v1/charges",
            auth=aiohttp.BasicAuth(self._stripe_key, ""),
            data={
                "amount": str(amount_cents),
                "currency": "usd",
                "customer": customer_id,
                "description": f"Billing {billing_period}",
            },
        ) as resp:
            body = await resp.json()
            result_future.complete([{"chargeId": body["id"], "customerId": customer_id}])

# In the Flink job graph:
# (with 100 concurrent requests, 100 wrong charges fire before first result returns)
# billed_stream = AsyncDataStream.unordered_wait(
#     events, AsyncBillingFunction(), timeout=10, time_unit=TimeUnit.SECONDS, capacity=100
# )

The fix is to issue one vault key per pipeline run — or per input batch, for bounded streams — before the async operator receives events. Vault keys issued by Keybrake carry a total dollar cap: the sum of all expected charges in the run plus a 10% buffer. Once the cumulative spend through the vault key reaches the cap, the proxy rejects further charges and returns a 402 error. In a batch billing run of 100 customers at $49.99 each, the vault key cap is set to $5,499 (100 × $49.99 × 1.10). A data error that doubles the amount would exhaust the cap after roughly 50 charges rather than hitting all 100 — stopping the bleeding at half the blast radius.

# async_billing.py — SAFE: per-pipeline vault key caps total spend, idempotency keys per event
from pyflink.datastream.functions import AsyncFunction, ResultFuture
import aiohttp
import hashlib
import json
import os

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:flink-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

async def issue_pipeline_vault_key(
    total_customers: int,
    amount_cents_each: int,
    billing_period: str,
) -> str:
    """Issue a vault key capped at total expected spend + 10%."""
    keybrake_api_key = os.environ["KEYBRAKE_API_KEY"]
    daily_usd_cap = int((total_customers * amount_cents_each * 1.1) / 100) + 1

    async with aiohttp.ClientSession() as session:
        async with session.post(
            "https://proxy.keybrake.com/vault/keys",
            headers={
                "Authorization": f"Bearer {keybrake_api_key}",
                "Content-Type": "application/json",
            },
            json={
                "vendor": "stripe",
                "daily_usd_cap": daily_usd_cap,
                "allowed_endpoints": ["POST /v1/charges"],
                "expires_in_seconds": 7200,
                "label": f"flink-billing/{billing_period}/{total_customers}customers",
            },
        ) as resp:
            data = await resp.json()
            return data["vault_key"]

class AsyncBillingFunction(AsyncFunction):
    """Async Stripe billing operator — uses per-pipeline vault key for spend-cap enforcement."""

    def open(self, runtime_context):
        # vault_key is set externally before the job starts (e.g., from job config)
        self._vault_key = os.environ["KEYBRAKE_VAULT_KEY"]
        self._session = None

    async def async_invoke(self, billing_event, result_future: ResultFuture):
        customer_id = billing_event["customerId"]
        amount_cents = billing_event["amountCents"]
        billing_period = billing_event["billingPeriod"]

        if self._session is None:
            self._session = aiohttp.ClientSession()

        idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

        async with self._session.post(
            "https://proxy.keybrake.com/stripe/v1/charges",
            auth=aiohttp.BasicAuth(self._vault_key, ""),
            headers={"Idempotency-Key": idempotency_key},
            data={
                "amount": str(amount_cents),
                "currency": "usd",
                "customer": customer_id,
                "description": f"Billing {billing_period}",
            },
        ) as resp:
            if resp.status == 402:
                # Vault key cap hit — surface as error record, do not raise
                result_future.complete([{
                    "customerId": customer_id,
                    "chargeId": None,
                    "status": "vault_cap_exceeded",
                    "billingPeriod": billing_period,
                }])
                return
            body = await resp.json()
            result_future.complete([{
                "customerId": customer_id,
                "chargeId": body.get("id"),
                "status": "charged",
                "billingPeriod": billing_period,
            }])

Failure mode 3: savepoint restore re-executes billing events processed since the savepoint

Flink savepoints are operator state snapshots taken deliberately — for job upgrades, parallelism rescaling, cluster migrations, or A/B topology changes. Unlike checkpoints (which are taken automatically for fault tolerance), savepoints are initiated by an operator action: a Flink CLI command, a REST API call, or a Kubernetes operator reconciliation. The savepoint captures operator state and source offsets at a point in time. When the job is restored from the savepoint — after the upgrade is applied, after the parallelism is changed — the Flink job reads from the source offset recorded in the savepoint and replays all events that arrived after the savepoint was taken.

In a continuously running billing Flink job — one that processes usage events from a Kafka topic and invokes Stripe for each — an operator-initiated savepoint taken for a routine job upgrade captures the Kafka consumer offset at that moment. The job is stopped, upgraded, and restarted from the savepoint. Flink's source operator seeks to the savepoint offset and replays every billing event that arrived after the savepoint was taken. If the upgrade takes 5 minutes and the billing stream processes 50 events per minute, 250 billing events are replayed through the BillingOperator. Without idempotency keys, each replayed event triggers a fresh stripe.charges.create() call. Stripe creates 250 duplicate charges — one for each customer whose billing event arrived during the upgrade window.

# The failure is structural, not in the operator code — but the operator's absence
# of idempotency keys is what makes the savepoint replay destructive.

# Safe savepoint restore requires two things:
# 1. Idempotency keys in every billing operator (prevents Stripe from creating
#    duplicate charges on replayed events)
# 2. An is-already-billed pre-flight check for event-time windows or aggregated
#    billing patterns where multiple source events map to one billing call.

# For simple per-event billing (one event = one charge), idempotency keys alone
# are sufficient: same event → same key → same Stripe charge object returned.

# For aggregated billing (N events → one charge per customer per window), the
# idempotency key must be derived from the window's defining parameters,
# not from any individual event:

def window_billing_idempotency_key(
    customer_id: str,
    amount_cents: int,
    window_start_ms: int,
    window_end_ms: int,
) -> str:
    """
    Key derived from window boundaries, not event content.
    Same window → same key → same Stripe charge object on savepoint replay.
    """
    raw = f"{customer_id}:{amount_cents}:{window_start_ms}:{window_end_ms}:flink-window-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

The structural protection is idempotency keys: if every billing call uses a key derived deterministically from the event's immutable fields (or, for windowed aggregations, from the window's time boundaries), savepoint replay produces no additional Stripe charges. The same records replay through the same operator, produce the same idempotency keys, and Stripe returns the original charge objects. The charges are already in the succeeded state; no new charges are created. Without this, every planned Flink maintenance operation — and every unplanned recovery — is a potential billing incident.

Additionally, configure a Flink DeduplicationFunction keyed on customer ID and billing period as a stage downstream of the billing operator. This function uses Flink's managed ValueState to record each billed (customer, period) pair. If a billing event for an already-billed customer arrives on replay — because the idempotency key check is a soft protection at the Stripe API layer, subject to the 24-hour rolling window — the deduplication function suppresses the downstream output and emits a SKIPPED_DUPLICATE record instead.

# dedup_function.py — Flink deduplication layer for billing operators
from pyflink.datastream import KeyedProcessFunction
from pyflink.datastream.state import ValueStateDescriptor
from pyflink.common.typeinfo import Types
import json

class BillingDeduplicationFunction(KeyedProcessFunction):
    """
    Keyed on (customerId, billingPeriod).
    Suppresses duplicate billing outputs from checkpoint/savepoint replay.
    """

    def open(self, runtime_context):
        descriptor = ValueStateDescriptor("billed", Types.BOOLEAN())
        self._billed_state = runtime_context.get_state(descriptor)

    def process_element(self, billing_result, ctx):
        if billing_result.get("status") != "charged":
            yield billing_result
            return

        already_billed = self._billed_state.value()
        if already_billed:
            # Suppress — this is a replay of an already-processed billing record
            yield {
                **billing_result,
                "status": "skipped_duplicate_replay",
            }
        else:
            self._billed_state.update(True)
            yield billing_result

Approach comparison

pytest enforcement suite

# test_flink_billing.py
import hashlib
import pytest

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:flink-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def window_billing_idempotency_key(
    customer_id: str,
    amount_cents: int,
    window_start_ms: int,
    window_end_ms: int,
) -> str:
    raw = f"{customer_id}:{amount_cents}:{window_start_ms}:{window_end_ms}:flink-window-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_idempotency_key_stable_across_checkpoint_replays():
    """Same billing event → same key across any number of Flink replays."""
    event = {"customerId": "cus_abc", "amountCents": 4999, "billingPeriod": "2026-07"}
    key1 = billing_idempotency_key(event["customerId"], event["amountCents"], event["billingPeriod"])
    key2 = billing_idempotency_key(event["customerId"], event["amountCents"], event["billingPeriod"])
    assert key1 == key2
    assert len(key1) == 32

def test_idempotency_key_differs_across_customers():
    key_a = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key_b = billing_idempotency_key("cus_xyz", 4999, "2026-07")
    assert key_a != key_b

def test_idempotency_key_differs_across_periods():
    key_jun = billing_idempotency_key("cus_abc", 4999, "2026-06")
    key_jul = billing_idempotency_key("cus_abc", 4999, "2026-07")
    assert key_jun != key_jul

def test_window_key_stable_across_savepoint_restore():
    """Same window boundaries → same key, even if restored from savepoint."""
    key1 = window_billing_idempotency_key("cus_abc", 4999, 1751328000000, 1753920000000)
    key2 = window_billing_idempotency_key("cus_abc", 4999, 1751328000000, 1753920000000)
    assert key1 == key2

def test_vault_key_not_bare_stripe_key(monkeypatch):
    monkeypatch.setenv("STRIPE_SECRET_KEY", "sk_live_real_secret")
    monkeypatch.setenv("KEYBRAKE_VAULT_KEY", "kb_vault_test_abc123")
    import os
    vault_key = os.environ["KEYBRAKE_VAULT_KEY"]
    assert vault_key.startswith("kb_"), "Flink billing operator must use vault key, not bare Stripe key"
    assert vault_key != os.environ["STRIPE_SECRET_KEY"]

Gap analysis

1. KeyedStream parallelism rescaling redistributes key groups during savepoint restore, causing the same (customerId, billingPeriod) to be processed by a different subtask with no access to prior state

When a Flink job is restored from a savepoint with a different parallelism than the original job, Flink redistributes key groups across the new number of subtasks. A KeyedStream operator that tracks billed customers in ValueState will have its state redistributed, but the redistribution is based on key hash mod new-parallelism — keys that were on subtask 0 may move to subtask 2. For newly added keys (customers whose billing events arrive after the savepoint), there is no state to redistribute: they start fresh with empty ValueState. For customers whose billing events straddle the savepoint (some events before, some after), the deduplication state from before the savepoint migrates correctly with the key group. The gap is customers whose billing events arrived just before the savepoint was taken — their charge completed, the state was set, but if the savepoint captured the state slightly before the operator processed the event, the state may be empty for that customer in the new job. Mitigate with idempotency keys as the primary safeguard (Stripe-layer protection) and the state-backed dedup as a secondary layer for cases outside the 24-hour Stripe window.

2. AsyncDataStream with orderedWait() serializes output but not API calls — all async requests still fire concurrently

A common misconception: using AsyncDataStream.orderedWait() instead of unorderedWait() serializes the order in which results are emitted to the downstream operator, but does not serialize the API calls themselves. Flink still dispatches up to capacity concurrent async_invoke() calls at any moment — the ordered vs unordered distinction only affects output ordering. A data error in the input stream produces the same blast radius regardless of which wait mode is used. Per-pipeline vault keys remain necessary for both modes.

3. Flink's exactly-once Kafka sink does not extend to external HTTP API calls

Flink's KafkaRecordSerializationSchema with two-phase commit (the FlinkKafkaProducer in older versions, KafkaSink with DeliveryGuarantee.EXACTLY_ONCE in newer) provides exactly-once delivery to Kafka by participating in Flink's checkpoint protocol. This guarantee applies to Flink-native sinks only. A billing ProcessFunction that calls stripe.charges.create() via HTTP is not a Flink-native sink and does not participate in two-phase commit. Flink cannot "roll back" a Stripe charge that completed before a checkpoint failure. The application-layer idempotency key is the only mechanism that makes the HTTP call idempotent across checkpoint replays — Flink's exactly-once semantics do not extend to it.

4. env.enableCheckpointing() with CheckpointingMode.AT_LEAST_ONCE increases replay volume

Flink's default checkpointing mode is EXACTLY_ONCE, which aligns checkpoint barriers across all parallel subtasks before taking the snapshot. AT_LEAST_ONCE mode skips barrier alignment and takes checkpoints faster, but can include records in the checkpoint that have already been processed by downstream operators — meaning those records will be re-processed (in addition to the standard gap between the checkpoint offset and the failure point). In a billing topology, AT_LEAST_ONCE checkpointing effectively doubles the potential replay window compared to EXACTLY_ONCE. Always use EXACTLY_ONCE checkpointing mode in billing Flink pipelines, and treat idempotency keys as mandatory regardless of mode.

FAQ

Does Flink's exactly-once checkpointing mean our billing calls are exactly-once?

No. Flink's exactly-once checkpointing guarantee applies to Flink's internal state backends and Flink-native sinks (Kafka, JDBC with supported connectors). HTTP API calls made inside operator functions are external side effects that Flink has no mechanism to roll back or deduplicate. The checkpoint protocol ensures that Flink's own state is consistent after recovery; it cannot ensure that an HTTP POST to Stripe was not already executed before the failure. Application-layer idempotency keys are required.

Should we set the idempotency key to the Kafka message offset?

No. Kafka message offsets are not stable across savepoint-triggered job restarts with parallelism changes: a record's offset in the source topic is fixed, but which Flink subtask processes it may change after rescaling. More importantly, for windowed aggregations where N source events produce one billing call, there is no single offset to use. Derive idempotency keys from the billing event's immutable business fields (customer ID, amount, period), not from Flink or Kafka infrastructure identifiers.

How long does Stripe honour an idempotency key?

Stripe retains idempotency key results for 24 hours from first use. If the same key is submitted within 24 hours, Stripe returns the original response object with no new charge. After 24 hours, the same key can be reused and will create a new charge. For Flink jobs with checkpoint intervals measured in minutes or hours, 24 hours is sufficient for fault tolerance recovery. For savepoint-based restores after multi-day maintenance windows, the 24-hour window may not be sufficient — pair idempotency keys with a database pre-flight check for cases where the restore lag exceeds 24 hours.

Can we use Flink's ValueState as the sole deduplication mechanism instead of idempotency keys?

Not safely. ValueState is checkpointed with the job, so it survives most task failures and checkpoint-based recovery. But ValueState does not prevent the billing call from being made — it only deduplicates the downstream output after the billing operator has already called Stripe. If the billing operator calls stripe.charges.create() and then fails before Flink can update the ValueState, the state is empty on recovery and the charge will fire again. Idempotency keys at the Stripe API layer are the primary safeguard; Flink state-backed deduplication is a complementary second layer.

Does the vault key need to be stored in Flink operator state for checkpoint recovery?

For per-pipeline vault keys (one key covering the entire billing run), the key should be passed as an environment variable or job parameter when the Flink job is started — it is set once and used by all parallel subtask instances. It does not need to be in Flink state; it lives in the operator's open() method. For per-event or per-customer vault keys, the key must arrive as part of the input record and should survive checkpoint replay because the input events themselves are replayed. Flink state is not the right place for vault keys.

What happens if the Keybrake proxy is unreachable when a billing event is processed?

The async operator receives an HTTP connection error from the proxy. The safest response is to treat this as a retriable error: raise an exception from async_invoke(), which causes Flink to fail the task and trigger checkpoint-based recovery. When the task restarts and replays the billing event, it retries the proxy call with the same idempotency key. Because the idempotency key is stable, if the first call reached Stripe before the proxy became unreachable, Stripe returns the original charge object on the second attempt with no duplicate charge. If the first call never reached Stripe (proxy was unreachable before forwarding), the second attempt creates the charge for the first time.

Related: Apache Beam Stripe Integration · Dagster Stripe Integration · Ray Stripe Integration · Prefect Stripe Integration · Temporal Stripe Integration