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

Apache Beam is a unified model for batch and streaming data processing, backed by runners including Google Cloud Dataflow, Apache Spark, and Apache Flink. Its execution model — retry failed elements, reassign failed bundles, redeliver unacked stream events — is designed for fault-tolerant large-scale computation. Those same recovery behaviors create specific billing hazards when Stripe API calls live inside a Beam pipeline: a DoFn that retries a failed element re-fires stripe.charges.create() whether or not the charge already succeeded; a worker failure replays an entire bundle of elements including ones that were already charged; and a streaming pipeline restores from checkpoint and reprocesses unacked PubSub messages that triggered billing on the previous run.

This post covers three Apache Beam-specific Stripe billing failure modes, the Python code that triggers each one, and the two-layer governance pattern — content-hash idempotency keys and per-pipeline-run vault keys via a spend-cap proxy — that eliminates all three without restructuring your pipeline.

Failure mode 1: DoFn retry re-executes process() from line 1 after stripe.charges.create() already succeeded

Beam's DoFn (transform function) abstraction processes one element at a time through its process() method. When process() raises an unhandled exception, the Beam runner marks the element as failed and retries it — typically after an exponential backoff — by calling process() again from the beginning. There is no mid-function checkpoint: if stripe.charges.create() returned a successful charge ID on line 14 and a subsequent BigQuery write raised a transient network error on line 22, the element is retried from line 1 and stripe.charges.create() is called again with identical arguments and no idempotency key.

# billing_pipeline.py — UNSAFE: DoFn retry re-fires stripe.charges.create() on write error
import apache_beam as beam
import stripe
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # unrestricted live key

class ChargeBillingDoFn(beam.DoFn):

    def process(self, element):
        customer_id = element["customer_id"]
        amount_cents = element["amount_cents"]
        billing_period = element["billing_period"]

        # DoFn retry restarts here — no idempotency key, no guard
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
        )
        # Transient BigQuery write error → element is retried → ch_B created
        write_to_bigquery(customer_id, charge.id, billing_period, amount_cents)
        yield {"customer_id": customer_id, "charge_id": charge.id}

def run_billing_pipeline(input_data, runner="DataflowRunner"):
    with beam.Pipeline(runner=runner) as p:
        (
            p
            | "ReadCustomers" >> beam.Create(input_data)
            | "ChargeCustomers" >> beam.ParDo(ChargeBillingDoFn())
            | "WriteResults" >> beam.io.WriteToBigQuery(...)
        )

On the retry, Stripe has no record of a previous call — no idempotency key was sent — so it treats the request as a new charge and creates ch_B for the same customer, same amount, same billing period. Dataflow's default retry budget is up to four attempts per element before the element is sent to the dead-letter sink. A single transient BigQuery outage can therefore produce up to four charges per customer in the same billing period. Dataflow's job logs show each retry as expected fault-tolerance behavior; the duplicate charges are invisible until a customer disputes a payment or a reconciliation job compares your BigQuery billing table to Stripe's charges export.

The fix: derive a content-hash idempotency key from the stable billing inputs — customer_id, amount_cents, and billing_period — and pass it with every stripe.charges.create() call. Beam guarantees that all retries of the same element receive the same input data, so the key is identical on every attempt. Stripe's idempotency layer returns the original ch_A on all subsequent calls without creating a new charge.

# billing_pipeline.py — SAFE: content-hash idempotency key survives all DoFn retries
import apache_beam as beam
import stripe
import hashlib
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

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

class ChargeBillingDoFn(beam.DoFn):

    def process(self, element):
        customer_id = element["customer_id"]
        amount_cents = element["amount_cents"]
        billing_period = element["billing_period"]

        idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

        # Same key on every retry — Stripe returns ch_A without creating ch_B, ch_C, ch_D
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
        write_to_bigquery(customer_id, charge.id, billing_period, amount_cents)
        yield {"customer_id": customer_id, "charge_id": charge.id}

Two additional improvements pair well with this: scope the Stripe key to POST /v1/charges only, so the DoFn cannot accidentally call refund or subscription endpoints regardless of what the pipeline input contains. And set a per-pipeline-run daily cap equal to the expected total charge volume — a unit confusion bug that passes amount_dollars where amount_cents is expected gets blocked by the proxy on the first call rather than creating a 100× charge that then retries up to four times.

Failure mode 2: bundle-level worker failure replays every element in the bundle including already-charged ones

Beam runners process pipeline data in groups of elements called bundles. Dataflow assigns each bundle to a specific worker VM and that worker processes all elements in the bundle sequentially. If the worker fails mid-bundle — due to OOM, preemption (common on Spot/Preemptible VMs), or a transient infrastructure failure — Dataflow reassigns the entire bundle to a new worker, which starts processing from the first element in the bundle regardless of how many elements the failed worker already processed.

# batch_billing.py — UNSAFE: bundle replay re-charges already-processed customers
import apache_beam as beam
import stripe
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # same key used across all workers

class BatchChargingDoFn(beam.DoFn):

    def process(self, element):
        customer_id = element["customer_id"]
        billing_period = element["billing_period"]

        # If the worker processing this bundle fails after some elements succeed,
        # the entire bundle is replayed on a new worker — no idempotency key,
        # so already-charged customers get ch_B on the replay
        charge = stripe.charges.create(
            amount=4999,
            currency="usd",
            customer=customer_id,
            description=f"Enterprise subscription {billing_period}",
        )
        yield {"customer_id": customer_id, "charge_id": charge.id, "status": charge.status}

The dangerous scenario is not edge-case behavior: Dataflow's default machine type for large pipelines includes Preemptible VMs, which can be reclaimed by Google Cloud at any time. A monthly billing pipeline that processes 10,000 customers in 50 bundles of 200 elements each has a non-trivial probability of at least one preemption during a full run. When a preempted worker's bundle is replayed, elements 1 through N (where N is however far the preempted worker got) receive a second charge — up to 200 duplicate charges per bundle failure. Dataflow's monitoring UI shows the bundle reassignment as expected fault-tolerance behavior with no visible signal that billing calls were duplicated.

The content-hash idempotency key closes this completely because the element data is identical on the replay. Stripe returns the existing ch_A for every element that was already charged on the failed worker. For additional protection, issue a per-pipeline-run vault key before the pipeline starts, scoped to POST /v1/charges with a spend cap equal to the total expected charge volume plus a buffer. A bug that causes the pipeline to run twice against the same input data exhausts the vault key's cap on the first run, and the proxy blocks all calls in the second run before any charges are created.

# batch_billing.py — SAFE: idempotency keys + per-run vault key cap
import apache_beam as beam
import stripe
import hashlib
import httpx
import os

KEYBRAKE_ADMIN_KEY = os.environ["KEYBRAKE_ADMIN_KEY"]
KEYBRAKE_PROXY_URL = "https://proxy.keybrake.com"

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

def issue_pipeline_vault_key(pipeline_run_id: str, customer_count: int, max_amount_cents: int) -> str:
    total_cap = customer_count * max_amount_cents
    resp = httpx.post(
        f"{KEYBRAKE_PROXY_URL}/keys",
        headers={"Authorization": f"Bearer {KEYBRAKE_ADMIN_KEY}"},
        json={
            "label": f"beam-run-{pipeline_run_id}",
            "vendor": "stripe",
            "daily_usd_cap": round(total_cap * 1.10 / 100, 2),
            "allowed_endpoints": ["POST /v1/charges"],
            "expires_in_seconds": 7200,  # 2-hour window for the pipeline run
        },
    )
    return resp.json()["vault_key"]

class BatchChargingDoFn(beam.DoFn):

    def __init__(self, vault_key: str, proxy_url: str):
        self.vault_key = vault_key
        self.proxy_url = proxy_url

    def setup(self):
        stripe.api_key = self.vault_key
        stripe.api_base = f"{self.proxy_url}/stripe"

    def process(self, element):
        customer_id = element["customer_id"]
        billing_period = element["billing_period"]
        idempotency_key = billing_idempotency_key(customer_id, 4999, billing_period)

        # Idempotency key: stable across bundle replay — Stripe returns ch_A on all attempts
        # Vault key cap: blocks a second full pipeline run from creating duplicate charges
        charge = stripe.charges.create(
            amount=4999,
            currency="usd",
            customer=customer_id,
            description=f"Enterprise subscription {billing_period}",
            idempotency_key=idempotency_key,
        )
        yield {"customer_id": customer_id, "charge_id": charge.id}

def run_billing_pipeline(pipeline_run_id: str, input_data: list):
    vault_key = issue_pipeline_vault_key(
        pipeline_run_id=pipeline_run_id,
        customer_count=len(input_data),
        max_amount_cents=4999,
    )
    with beam.Pipeline(runner="DataflowRunner") as p:
        (
            p
            | "ReadCustomers" >> beam.Create(input_data)
            | "ChargeCustomers" >> beam.ParDo(BatchChargingDoFn(vault_key, KEYBRAKE_PROXY_URL))
            | "WriteResults" >> beam.io.WriteToBigQuery(...)
        )

Failure mode 3: streaming pipeline PubSub redelivery fires Stripe calls a second time after checkpoint failure

In streaming mode with a PubSub source, Beam processes messages and acknowledges them to PubSub after the downstream sink write completes. This is the source of Beam's at-least-once delivery guarantee: messages are not acked until there is proof that the downstream write succeeded. If the pipeline crashes between a Stripe call and the BigQuery sink write — due to an OOM kill, a driver failure, or a Dataflow job preemption — the message was never acked. PubSub redelivers it when the pipeline restarts from its last checkpoint. The redelivered message triggers the same billing event, and without an idempotency key, stripe.charges.create() creates a new charge for the same customer.

# streaming_billing.py — UNSAFE: PubSub redelivery triggers duplicate charge on pipeline restart
import apache_beam as beam
from apache_beam.io import ReadFromPubSub
from apache_beam.transforms.window import FixedWindows
import stripe
import json
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

class ProcessBillingEventDoFn(beam.DoFn):

    def process(self, element, window=beam.DoFn.WindowParam):
        event = json.loads(element.decode("utf-8"))
        customer_id = event["customer_id"]
        amount_cents = event["amount_cents"]
        billing_period = event["billing_period"]

        # Pipeline crash after this line but before BigQuery write
        # → PubSub redelivers the message on restart
        # → process() fires again → ch_B created
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Usage charge {billing_period}",
        )
        # Pipeline crash here: BigQuery write fails, PubSub does not ack, message redelivered
        yield {"customer_id": customer_id, "charge_id": charge.id}

def run_streaming_pipeline():
    with beam.Pipeline(runner="DataflowRunner") as p:
        (
            p
            | "ReadEvents" >> ReadFromPubSub(topic="projects/myproject/topics/billing-events")
            | "WindowInto" >> beam.WindowInto(FixedWindows(60))
            | "ChargeCustomers" >> beam.ParDo(ProcessBillingEventDoFn())
            | "WriteToBQ" >> beam.io.WriteToBigQuery(...)
        )

The streaming failure mode is especially difficult to observe because it requires two conditions to coincide: a pipeline restart event and an in-flight billing message at that exact moment. Dataflow Streaming jobs restart automatically after worker failures without operator intervention, making this failure mode silent. The redelivered PubSub message carries no signal that it was already partially processed; it looks identical to a new billing event. The resulting duplicate charge appears as a separate Stripe object with a different charge.id and a different request_id header — no obvious signal that it duplicates an earlier charge for the same customer and billing period.

Content-hash idempotency keys derived from customer_id, amount_cents, and billing_period embedded in the PubSub message close this completely. The redelivered message carries the same payload, which produces the same idempotency key, and Stripe returns the original ch_A from its idempotency store without creating a new charge. A stable billing event identifier from the PubSub message — for example, a billing_event_id field set by the upstream publisher — can also serve as the idempotency key directly, as long as the publisher guarantees it is stable across PubSub redeliveries (which it is: PubSub preserves the message data and attributes on redelivery).

# streaming_billing.py — SAFE: content-hash idempotency key survives PubSub redelivery
import apache_beam as beam
from apache_beam.io import ReadFromPubSub
from apache_beam.transforms.window import FixedWindows
import stripe
import hashlib
import json
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

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

class ProcessBillingEventDoFn(beam.DoFn):

    def process(self, element, window=beam.DoFn.WindowParam):
        event = json.loads(element.decode("utf-8"))
        customer_id = event["customer_id"]
        amount_cents = event["amount_cents"]
        billing_period = event["billing_period"]

        idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

        # Same key on PubSub redelivery — Stripe returns ch_A without creating ch_B
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description=f"Usage charge {billing_period}",
            idempotency_key=idempotency_key,
        )
        yield {"customer_id": customer_id, "charge_id": charge.id}

Approach comparison

Gap analysis: four more Apache Beam billing risks

1. Dataflow Spot VM preemption creates a retry layer independent of DoFn retry configuration

Beam's DoFn retry configuration controls what happens when process() raises a Python exception. VM preemption is a different failure mode: the entire worker process is terminated by the cloud provider with no opportunity to raise an exception or run cleanup code. The bundle assigned to the preempted worker is reassigned to another worker and replayed from the beginning — this is a runner-level retry that operates entirely outside your DoFn's exception handling. Content-hash idempotency keys are the only protection, because the runner does not invoke any DoFn error handling during a preemption-triggered bundle reassignment. Setting --use_public_ips=false or switching to non-preemptible VMs reduces frequency but does not eliminate preemption for long-running streaming jobs.

2. beam.GroupByKey does not deduplicate — multiple messages with the same customer key each produce an independent Stripe call

A common pattern is to use beam.GroupByKey to aggregate events by customer before billing, then map over the groups. GroupByKey collects all values for a given key into an iterable, but it does not deduplicate values — if the same billing event appears twice in the input (from a PubSub redelivery that occurred before the GroupByKey stage), both copies arrive in the iterable for that customer. A DoFn that iterates over the group and calls stripe.charges.create() for each value in the iterable will fire two charges for the customer. Idempotency keys derived from the stable event content close this; grouping by key does not.

3. Side input refreshes during long-running streaming jobs can cause re-processing of elements whose side input value changed

Beam's side inputs allow DoFns to read from a slowly changing dataset — for example, a pricing table that maps subscription tiers to charge amounts. When a side input is refreshed (because the underlying dataset was updated), Beam may need to re-process elements in the current window using the new side input values. If a billing DoFn uses a pricing side input and the price changes mid-window, Beam may re-invoke process() for elements that were already processed with the old price. Without an idempotency key, this creates a second charge for the same customer in the same billing window. The idempotency key must be derived from the element data only — not from the side input value — so it remains stable across re-processing triggered by a side input refresh.

4. Cross-language transforms introduce additional retry boundaries that bypass Python DoFn error handling

Beam's cross-language transform support allows Python pipelines to invoke Java or Go transforms via the expansion service protocol. If your billing pipeline uses a cross-language transform for a step downstream of the Stripe call — for example, a Java-based BigQuery sink — and that transform fails, the failure propagates back to the Python runner, which retries the affected elements. The Python DoFn's process() method runs again from the beginning for each retried element, re-firing any Stripe calls that preceded the cross-language step. The retry boundary here is at the element level, not at the individual transform level, so idempotency keys in the Python DoFn protect all downstream cross-language failure modes as long as they are applied before the first Stripe call.

Enforcement with pytest

# test_beam_billing.py
import hashlib
import pytest
from unittest.mock import patch, MagicMock

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

def test_idempotency_key_stable_across_dofn_retries():
    """Same element inputs always produce the same key regardless of retry count."""
    key_attempt_1 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key_attempt_2 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key_attempt_3 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    assert key_attempt_1 == key_attempt_2 == key_attempt_3

def test_idempotency_key_distinct_per_billing_period():
    """Different billing periods produce different keys — no cross-period dedup."""
    key_july = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key_august = billing_idempotency_key("cus_abc", 4999, "2026-08")
    assert key_july != key_august

def test_idempotency_key_distinct_per_customer():
    """Different customers in the same bundle get independent keys."""
    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_vault_key_rejects_refund_endpoint():
    """Vault key scoped to POST /v1/charges cannot call POST /v1/refunds."""
    with patch("stripe.Refund.create") as mock_refund:
        mock_refund.side_effect = Exception(
            "403 Forbidden: vault key not authorized for POST /v1/refunds"
        )
        with pytest.raises(Exception, match="not authorized"):
            import stripe
            stripe.Refund.create(charge="ch_test_abc")

def test_preflight_audit_check_blocks_duplicate_pipeline_run():
    """Pre-flight check short-circuits billing if charge already recorded for period."""
    mock_db = MagicMock()
    mock_db.get_charge.return_value = {"charge_id": "ch_A", "status": "succeeded"}

    def preflight_check(customer_id, billing_period):
        existing = mock_db.get_charge(customer_id, billing_period)
        if existing:
            return existing["charge_id"]
        return None

    result = preflight_check("cus_abc", "2026-07")
    assert result == "ch_A"
    mock_db.get_charge.assert_called_once_with("cus_abc", "2026-07")

FAQ

Is the Beam element's timestamp or window ID safe to use as an idempotency key?

No. Beam's element timestamps and window assignments are runner-managed metadata that can shift during windowing re-assignments, late-data handling, or pipeline graph optimizations. A window boundary change causes Beam to reassign elements to different windows with different IDs, which would generate a different idempotency key for the same billing event — causing the event to create a new charge rather than returning the existing one. The content-hash derived from your billing inputs — customer_id, amount_cents, billing_period — is stable because it depends only on the element's data payload, not on Beam's internal execution metadata.

Does Dataflow's exactly-once guarantee eliminate the need for idempotency keys?

Dataflow offers exactly-once semantics for sink writes via its shuffle and state layers, but this guarantee applies to the pipeline's output records — it does not extend to side effects that your DoFn produces, including external API calls like stripe.charges.create(). Your DoFn's process() method can be invoked multiple times for the same element even when the downstream BigQuery write is delivered exactly once. Idempotency keys are required at the Stripe API layer regardless of the runner's delivery guarantees for its own output.

Can vault keys be issued inside a DoFn's setup() method, or must they be issued before the pipeline starts?

You can issue the vault key in setup(), which runs once per DoFn instance per worker. For batch pipelines, issuing the key before the pipeline starts (as shown in Failure mode 2) is preferable because it allows you to set the spend cap based on the full input size. If you issue the key in setup(), each worker gets its own vault key with a cap sized for that worker's share of the input — which provides isolation but requires you to know the per-worker element count in advance. For streaming pipelines, setup() is the practical choice because the pipeline runs indefinitely; issue a new key with a rolling 24-hour spend cap and refresh it in setup() when the key is near expiry.

What happens if the Keybrake proxy is unreachable when a Beam worker calls stripe.charges.create()?

The Stripe call fails with a connection error rather than silently bypassing the proxy. Beam treats this as a DoFn exception and retries the element according to your retry configuration. Because the idempotency key is derived from stable inputs, the retry against the proxy (once it recovers) produces the same key, and Stripe returns the existing charge if it was created on a prior attempt. No duplicate charge is created. Configure your Dataflow job's retry budget to allow sufficient retries to cover a proxy recovery window — the default of four attempts with exponential backoff is typically sufficient for transient proxy unavailability.

How does Dataflow Flex Templates handle secrets rotation for the Stripe key and Keybrake admin key?

Dataflow Flex Templates inject secrets as environment variables at job launch time, sourced from Secret Manager via the --env_var flag or directly in the template spec. A secrets rotation event (updating the Keybrake admin key in Secret Manager) does not affect the running job — workers already have the old key in memory. New jobs launched after the rotation pick up the new key. For long-running streaming jobs, build a key rotation mechanism into your DoFn that periodically fetches a fresh admin key from Secret Manager and issues a new vault key, rather than relying on the environment variable that was set at job launch.

Should I use Beam batch or streaming mode for subscription billing workloads?

Batch mode (Dataflow batch jobs) is the right choice for subscription billing — monthly or usage-based charges that run on a schedule. Batch pipelines have a defined start and end, which makes it practical to issue a single vault key before the run starts and set its spend cap based on the full input size. Streaming mode introduces the PubSub redelivery failure mode (covered above) and makes spend cap sizing more complex because the input is unbounded. Reserve streaming mode for event-driven billing — real-time usage metering, per-call charges, or webhook-triggered billing — where the latency advantage justifies the additional idempotency complexity.

Related: BentoML · Metaflow · ZenML · Flyte · Ray · Dagster · Prefect · Stripe restricted key permissions reference