Databricks Jobs Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Databricks Jobs is the managed workflow orchestrator for notebooks, Python scripts, JAR files, and dbt projects running on Databricks clusters. Teams reach for it when billing pipelines need the full power of a Spark cluster — aggregating usage data from Delta Lake, computing charges per customer tier, then calling Stripe in a final billing task. But three Databricks behaviors — notebook task retries that restart the entire notebook from cell 1, for_each_task iterating over an input list that can contain duplicates, and max_concurrent_runs allowing a second billing run to start while the first is still executing — introduce Stripe billing failure modes that Databricks' own observability and Stripe's restricted key feature do not surface on their own.

This post covers all three failure modes with Databricks Jobs YAML and Python billing code, and the two-layer governance pattern — content-hash idempotency keys plus per-task vault keys via a spend-cap proxy — that eliminates all three without restructuring your Databricks workflow.

Failure mode 1: max_retries re-runs the entire notebook from cell 1 after a successful charge and a downstream Delta Lake write failure

Databricks Jobs task retry is configured via max_retries in the task definition. When a notebook task fails — any cell raises an exception that is not caught — Databricks marks the task run as failed and, if max_retries > 0, schedules a new task attempt. The retry re-executes the entire notebook from cell 1 on a fresh cluster context. The new run has no awareness of which cells the previous attempt executed or which API calls it completed successfully; it starts from scratch with the same parameters and the same secrets injected from Databricks Secrets.

The failure mode is specific to billing notebooks that call stripe.charges.create() partway through and then write a billing record to Delta Lake afterward. Cell 4 calls Stripe and returns charge object ch_A. Cell 8 attempts to write the billing record to a Delta Lake table using spark.sql("INSERT INTO billing.charges VALUES ..."). If cell 8 raises a schema mismatch (AnalysisException: cannot resolve column 'charge_metadata') or a permission error (SecurityException: User does not have MODIFY privilege on billing.charges), the notebook exits with an exception. Databricks marks the task failed. With max_retries: 2, a new run starts from cell 1. Cell 4 calls stripe.charges.create() again for the same customer — without an idempotency key — and creates charge ch_B. The customer's bank sees two charges for the same billing period.

# billing_notebook.py (Databricks notebook cell 4) — UNSAFE: no idempotency key
import stripe
import os

stripe.api_key = dbutils.secrets.get(scope="stripe", key="secret_key")  # full-access key

# Cell 4: charge the customer
# UNSAFE: no idempotency_key.
# If cell 8 (Delta Lake write) fails after this call succeeds,
# Databricks retries the entire notebook from cell 1.
# Cell 4 runs again → stripe.charges.create() fires again → ch_B created.
charge = stripe.Charge.create(
    amount=amount_cents,
    currency="usd",
    customer=customer_id,
    description=f"Usage billing {billing_period}",
)

print(f"Charged {customer_id}: {charge['id']}")

# ...cells 5-7: aggregate usage, compute line items...

# Cell 8: write billing record to Delta Lake — UNSAFE: if this fails, retry re-runs cell 4
spark.sql(f"""
    INSERT INTO billing.charges (customer_id, billing_period, charge_id, amount_cents)
    VALUES ('{customer_id}', '{billing_period}', '{charge['id']}', {amount_cents})
""")

The fix has two parts. First, compute a content-hash idempotency key from the billing inputs before calling Stripe. The key must be derived from values that are stable across all notebook retry attempts: sha256(customer_id:amount_cents:billing_period:databricks-billing)[:32]. Do not use the Databricks run ID or task attempt number — these change between retries and will produce a different key on each attempt, creating a new charge instead of deduplicating. Second, add a pre-flight check at the start of the notebook: query the Delta Lake billing table for a record with the same customer and billing period before reaching the Stripe call cell. If the record exists, skip the Stripe cell entirely. The pre-flight check protects beyond Stripe's 24-hour idempotency window and is the critical guard when a schema migration leaves an orphaned billing record from a prior run.

# billing_notebook.py — SAFE: content-hash idempotency key + pre-flight check + vault key
import stripe
import hashlib
import os

KEYBRAKE_PROXY_URL = os.environ.get("KEYBRAKE_PROXY_URL", "https://proxy.keybrake.com")
stripe.api_key = dbutils.secrets.get(scope="keybrake", key="vault_key")  # scoped vault key
stripe.api_base = f"{KEYBRAKE_PROXY_URL}/stripe"

# Cell 1: pre-flight check — skip if already billed (protects beyond Stripe's 24h window)
already_billed = spark.sql(f"""
    SELECT COUNT(*) as cnt FROM billing.charges
    WHERE customer_id = '{customer_id}'
      AND billing_period = '{billing_period}'
""").collect()[0]["cnt"]

if already_billed > 0:
    print(f"Skipping {customer_id} — already billed for {billing_period}")
    dbutils.notebook.exit("already_billed")

# Cell 4: charge the customer with a stable content-hash idempotency key
def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:databricks-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

idempotency_key = billing_idempotency_key(customer_id, amount_cents, billing_period)

try:
    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=customer_id,
        description=f"Usage billing {billing_period}",
        idempotency_key=idempotency_key,
    )
except stripe.error.CardError as e:
    print(f"Card declined for {customer_id}: {e.user_message}")
    dbutils.notebook.exit("card_declined")  # exit cleanly — not an error, no retry
except stripe.error.InvalidRequestError as e:
    print(f"Invalid request for {customer_id}: {e}")
    dbutils.notebook.exit("invalid_request")  # exit cleanly — permanent error, no retry

# Cell 8: write billing record to Delta Lake
spark.sql(f"""
    INSERT INTO billing.charges (customer_id, billing_period, charge_id, amount_cents)
    VALUES ('{customer_id}', '{billing_period}', '{charge['id']}', {amount_cents})
    ON DUPLICATE KEY UPDATE charge_id = VALUES(charge_id)
""")

Failure mode 2: for_each_task with a duplicate in the input list dispatches two concurrent billing instances for the same customer

The Databricks Jobs for_each_task construct (available in the Jobs API 2.1 / UI since DBR 13.0) iterates over an input list and dispatches a concurrent task instance for each item. Teams use it for fan-out billing: the input list is a JSON array of customer IDs or cohort identifiers, and the child task bills each customer independently. The concurrency setting — for_each_task.concurrency — controls how many child instances run simultaneously. With concurrency: 10 and 500 customers, 10 instances run at a time, each receiving a different customer ID from the input list.

The failure mode emerges from data quality issues upstream. The input list is typically constructed by a preceding task that reads from Delta Lake: a SQL query produces the list of customers to bill for the period. If the upstream query contains a bug — a UNION without DISTINCT, a join that produces fan-out, or a deduplication step that was accidentally removed during a refactor — the input list can contain the same customer ID twice. The for_each_task does not validate uniqueness of its input; it dispatches one task instance per list item regardless of duplicates. Two concurrent child instances each receive customer_id: cus_abc, each independently call stripe.charges.create() with no cross-instance coordination, and both charges succeed. The customer is billed twice for the same period.

# databricks_job.yml — UNSAFE: for_each_task with no input dedup guard
tasks:
  - task_key: build_billing_list
    notebook_task:
      notebook_path: /billing/build_customer_list
    new_cluster:
      spark_version: "15.4.x-scala2.12"
      node_type_id: i3.xlarge
      num_workers: 2

  - task_key: bill_customers
    depends_on:
      - task_key: build_billing_list
    for_each_task:
      # UNSAFE: input list from upstream task may contain duplicates.
      # for_each_task dispatches one instance per item, including duplicate customer IDs.
      inputs: "{{tasks.build_billing_list.values.customer_list}}"
      concurrency: 10
      task:
        task_key: bill_single_customer
        notebook_task:
          notebook_path: /billing/bill_customer
          base_parameters:
            customer_id: "{{input}}"
            billing_period: "2026-07"

The fix operates at two levels. First, deduplicate the input list in the upstream task before passing it to for_each_task: use SELECT DISTINCT customer_id FROM billing_queue and add an assertion that the output list has no duplicates before writing the task value. A data quality check that fails fast is far preferable to discovering duplicate charges from a customer support ticket. Second, use the content-hash idempotency key inside the child billing notebook: even if two concurrent instances receive the same customer ID, the Stripe idempotency key derived from sha256(customer_id:amount_cents:billing_period) is identical for both — Stripe deduplicates the second charge and returns the first charge object. The pre-flight Delta Lake check inside each child task catches duplicates that slip through past the 24-hour idempotency window.

# build_customer_list_notebook.py — SAFE: deduplicate before for_each_task
from pyspark.sql.functions import col, countDistinct

# Build the billing list with explicit deduplication
billing_df = spark.sql("""
    SELECT DISTINCT customer_id, amount_cents
    FROM billing.usage_summary
    WHERE billing_period = '2026-07'
      AND payment_status = 'pending'
""")

# Assert no duplicates before writing the task value (fail fast)
total = billing_df.count()
distinct = billing_df.select("customer_id").distinct().count()

if total != distinct:
    raise ValueError(
        f"Duplicate customer_ids in billing list: {total} rows, {distinct} distinct. "
        "Check upstream JOIN logic before billing."
    )

# Write deduplicated list as task value for for_each_task
customer_list = [row["customer_id"] for row in billing_df.collect()]
dbutils.jobs.taskValues.set(key="customer_list", value=customer_list)
print(f"Billing {len(customer_list)} customers for 2026-07 (no duplicates confirmed)")

Failure mode 3: max_concurrent_runs allows a second billing run to start while the first is still executing

Databricks Jobs max_concurrent_runs controls how many active runs of a job can exist simultaneously. The default is 1, which prevents concurrent runs. Teams sometimes increase it above 1 to allow testing runs to co-exist with scheduled production runs, or to allow parallel runs for different billing periods. The failure mode occurs when the billing job has max_concurrent_runs: 2 (or higher) and the billing run duration exceeds the schedule interval: a monthly billing job runs on the first of each month, but a large cohort causes one run to take 75 minutes; when the next scheduled window fires (for example, a retry trigger or a second schedule expression), Databricks starts a second run while the first is still executing. Both runs process the same customer cohort for the same billing period, both independently call stripe.charges.create(), and both succeed.

This failure mode is subtle because the two runs have different Databricks run IDs and appear as separate entries in the job run history. Unless an engineer notices that two runs for the same job completed on the same day, the duplication is invisible in the Databricks UI. It surfaces only when customers report being charged twice, or when a billing reconciliation script compares Stripe charges against the Delta Lake billing table and finds two charge records for the same customer and period.

# databricks_job.yml — UNSAFE: max_concurrent_runs > 1 allows overlapping billing runs
name: monthly-billing
max_concurrent_runs: 2  # UNSAFE: allows two billing runs for the same period simultaneously

schedule:
  quartz_cron_expression: "0 0 1 * *"  # 1st of each month at midnight UTC
  timezone_id: "UTC"
  pause_status: UNPAUSED

tasks:
  - task_key: run_billing
    notebook_task:
      notebook_path: /billing/monthly_billing
      base_parameters:
        billing_period: "{{job.start_time.strftime('%Y-%m')}}"
    new_cluster:
      spark_version: "15.4.x-scala2.12"
      node_type_id: i3.2xlarge
      num_workers: 4

The fix is to set max_concurrent_runs: 1 for all billing jobs without exception. There is no legitimate reason to allow two runs of the same billing job to execute concurrently for the same billing period. If testing alongside production is required, use a separate job with a test billing period parameter. Additionally, use a Delta Lake distributed lock inside the billing notebook: at notebook startup, attempt to insert a row into a billing_locks table with WHEN NOT MATCHED; if the insert succeeds, the current run holds the lock; if it fails (because the row already exists from a concurrent run), call dbutils.notebook.exit("lock_held"). The content-hash idempotency key and pre-flight check are the final backstop even if two runs do reach the Stripe call simultaneously.

# databricks_job.yml — SAFE: max_concurrent_runs: 1 + distributed lock in notebook
name: monthly-billing
max_concurrent_runs: 1  # SAFE: prevents overlapping billing runs at the scheduler level

schedule:
  quartz_cron_expression: "0 0 1 * *"
  timezone_id: "UTC"
  pause_status: UNPAUSED

tasks:
  - task_key: run_billing
    notebook_task:
      notebook_path: /billing/monthly_billing
      base_parameters:
        billing_period: "{{job.start_time.strftime('%Y-%m')}}"
    new_cluster:
      spark_version: "15.4.x-scala2.12"
      node_type_id: i3.2xlarge
      num_workers: 4
    max_retries: 1
    retry_on_timeout: false

# billing_notebook.py (cell 1) — SAFE: distributed lock via Delta Lake MERGE
from delta.tables import DeltaTable
from pyspark.sql import Row
from datetime import datetime, timezone

# Attempt to acquire a distributed billing lock for this period.
# Uses Delta Lake MERGE to atomically insert the lock row only if it doesn't exist.
lock_df = spark.createDataFrame([Row(
    billing_period=billing_period,
    locked_by=dbutils.entry_point.getDbutils().notebook().getContext().currentRunId().get(),
    locked_at=datetime.now(timezone.utc).isoformat(),
)])

lock_table = DeltaTable.forName(spark, "billing.locks")
result = (
    lock_table.alias("existing")
    .merge(lock_df.alias("new"), "existing.billing_period = new.billing_period")
    .whenNotMatchedInsertAll()
    .execute()
)

# If no rows were inserted, another run holds the lock for this period
rows_inserted = result.filter("metrics.numTargetRowsInserted > 0").count()
if rows_inserted == 0:
    print(f"Billing lock for {billing_period} held by another run — exiting")
    dbutils.notebook.exit("lock_held")  # exit cleanly, no retry

print(f"Acquired billing lock for {billing_period}")

Approach comparison

Gap analysis: four scenarios the above does not cover

1. Databricks Workflow run_if: ALL_DONE triggering the billing task even when the upstream data-prep task failed, combined with task-level max_retries, produces billing runs against incomplete or stale input data. When a billing task is configured with run_if: ALL_DONE (run regardless of upstream task success), and the upstream data-prep task fails partway through building the customer list, the billing task starts with whatever partial output the data-prep task wrote before failing. If the billing task also has max_retries: 2 and its own run fails on the incomplete data, the retry picks up the same incomplete list. The result is a partial billing run — some customers charged, others skipped — with no automatic reconciliation. Use run_if: ALL_SUCCESS for billing tasks that depend on upstream data prep, and treat billing task failures as blocking rather than as conditions to retry around. The pre-flight check will protect against double-charging on retry, but cannot add missing customers from an incomplete upstream list.

2. Databricks Jobs parameter passing via {{job.start_time}} computes the billing period at schedule fire time, not at the actual billing logic execution time, creating off-by-one period errors near month boundaries. A billing job scheduled at 0 0 1 * * (first of the month at midnight UTC) uses {{job.start_time.strftime('%Y-%m')}} to determine the billing period. If the Databricks scheduler fires the job at 23:59:58 UTC on the last day of the month and the job parameter is evaluated at fire time rather than at task execution start, the billing period is computed as the prior month. The content-hash idempotency key is derived from the computed billing period — if two runs compute different periods for the same billing cycle (one as 2026-06, one as 2026-07), the idempotency keys are different and neither protects against duplication. Pass billing period as an explicit parameter from an upstream task that asserts the correct period, rather than relying on schedule-time substitution near boundary conditions.

3. Vault key TTL expires during long-running Spark billing aggregation that precedes the Stripe calls. A common Databricks billing workflow reads usage data from Delta Lake (5–15 minutes), aggregates it into per-customer charge amounts (10–30 minutes on large clusters), and only then calls Stripe for each customer. If the vault key is issued at the start of the notebook with a TTL sized to the expected Stripe-call portion of the run, but not to the total notebook duration, the key expires before the billing cells are reached. All Stripe calls through the proxy return HTTP 401. The notebook exits with an authentication error. With max_retries: 1, Databricks runs the notebook again from cell 1 — the data aggregation re-runs and the Stripe calls fail again with the same expired key. Issue vault keys with TTL set to total_expected_notebook_duration + 30 minutes, or issue the vault key in a dedicated cell immediately before the Stripe call section rather than at notebook startup.

4. Databricks Jobs multi-task workflow with task_key dependencies does not propagate task parameter changes when retrying a single failed task. In a multi-task workflow, Databricks can retry a single failed downstream task without re-running upstream tasks. If the upstream task wrote task values that include a billing period or cohort snapshot, and those values are stale by the time the downstream billing task retries (for example, because a separate job modified the underlying Delta table between the original run and the retry), the billing task may charge customers based on an outdated cohort snapshot. For billing workflows with multi-task dependencies, always read the customer list and charge amounts directly from Delta Lake at billing task execution time — not from upstream task values — to ensure the billing task operates on current data regardless of when the retry fires.

Enforcement tests

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

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

def test_idempotency_key_is_stable_across_notebook_retries():
    """Same inputs on any retry attempt must produce the same key."""
    key_attempt_1 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    key_attempt_2 = billing_idempotency_key("cus_abc", 4999, "2026-07")
    assert key_attempt_1 == key_attempt_2
    assert len(key_attempt_1) == 32

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

def test_preflight_check_skips_already_billed_customer(mocker):
    """Delta Lake pre-flight must skip customers with existing billing records."""
    mock_spark = MagicMock()
    mock_result = MagicMock()
    mock_result.collect.return_value = [{"cnt": 1}]
    mock_spark.sql.return_value = mock_result
    mocker.patch("billing_notebook.spark", mock_spark)
    with patch("billing_notebook.stripe.Charge.create") as mock_charge:
        from billing_notebook import check_and_bill
        result = check_and_bill(mock_spark, "cus_abc", 4999, "2026-07")
        assert result == "already_billed"
        mock_charge.assert_not_called()

def test_for_each_input_dedup_raises_on_duplicate_customer_ids():
    """build_customer_list must raise if duplicate customer_ids are detected."""
    from pyspark.sql import Row
    mock_spark = MagicMock()
    all_rows = [Row(customer_id="cus_abc"), Row(customer_id="cus_abc")]  # duplicate
    distinct_rows = [Row(customer_id="cus_abc")]
    mock_spark.sql.return_value.count.side_effect = [2, 1]  # total=2, distinct=1
    with pytest.raises(ValueError, match="Duplicate customer_ids"):
        total = mock_spark.sql.return_value.count()
        distinct = mock_spark.sql.return_value.count()
        if total != distinct:
            raise ValueError(f"Duplicate customer_ids in billing list: {total} rows, {distinct} distinct.")

def test_card_error_exits_cleanly_without_raising():
    """CardError must be caught and not re-raised to trigger Databricks notebook retry."""
    try:
        raise stripe.error.CardError("Card declined", "card_declined", "card_declined", 402)
    except stripe.error.CardError:
        pass  # correct: caught, notebook calls dbutils.notebook.exit("card_declined")

Frequently asked questions

Can I use the Databricks run ID as the idempotency key instead of a content hash?

No. The Databricks run ID — available as dbutils.entry_point.getDbutils().notebook().getContext().currentRunId().get() — is different for each run attempt, including retries. With max_retries: 2, the original run has run ID 12345 and the retry run has run ID 12346. Using the run ID as the idempotency key means the retry's stripe.charges.create() call uses a different key than the original run, so Stripe treats it as a new charge and creates a duplicate. Content-hash idempotency keys derived from billing inputs (customer_id:amount_cents:billing_period) are stable regardless of which run attempt executes the billing call. See also: idempotency key patterns for agent workflows.

Does the Databricks for_each_task have a native deduplication option for the input list?

No. The for_each_task dispatches one task instance per item in the input list without any built-in uniqueness enforcement. If the input list contains ["cus_abc", "cus_abc", "cus_xyz"], three task instances are dispatched and two will attempt to bill cus_abc. Deduplication must happen in the upstream task that constructs the input list — use SELECT DISTINCT and assert count equality before writing the task value. For production billing workflows, also add a data quality check that raises an alert (and fails the task) if duplicates are detected, so the issue is caught before billing instances run.

How should vault keys be issued per for_each_task child instance?

The cleanest pattern is to issue vault keys in the upstream list-building task: for each customer in the billing list, call the Keybrake API to issue a vault key scoped to that customer's expected charge, then include the vault key alongside the customer ID in the for_each_task input list as a JSON object: {"customer_id": "cus_abc", "vault_key": "vault_key_xxx", "amount_cents": 4999}. Each child task reads its vault key from the input item and uses it for the Stripe call. This gives each child a separate key with a per-customer spend cap, so a calculation error for one customer's charge cannot propagate beyond that customer's vault key limit. If the input list approach makes the payload too large, store the vault key map in a Delta table keyed by customer ID and billing period, and have each child task read its key at startup.

What happens when a Databricks Job is paused and then unpaused after missed schedule windows?

By default, Databricks Jobs does not fire catch-up runs for missed schedule windows when a job is unpaused. Only the next scheduled window fires after unpausing. This is unlike some schedulers (Airflow with catchup=True, Celery Beat on restart) that fire backlogged runs. However, if max_concurrent_runs was set above 1 before pausing, and a run from before the pause is still in RUNNING state when the job is unpaused and the next schedule fires, those two runs will overlap. Always verify that no runs are in flight before unpausing a billing job, and set max_concurrent_runs: 1 unconditionally.

Can Keybrake vault keys be issued per for_each_task instance from within the task itself?

Yes, but this introduces a risk: if two for_each_task instances receive the same customer ID (from a duplicate in the input list), both instances will call the Keybrake API to issue separate vault keys — and both keys will be independent, each with its own spend cap. The two instances can then both call stripe.charges.create() through their separate vault keys, and each key's cap is not exceeded, so neither is blocked. The content-hash idempotency key handles the Stripe-level dedup (within 24 hours), but the vault-key-per-instance approach does not prevent the second API call from reaching the proxy. Issue vault keys in the upstream task per unique customer ID and include them in the deduplicated input list to ensure no two instances receive the same customer-vault-key pair.

How does Keybrake's audit log help with Databricks billing investigations?

The Keybrake audit log records every Stripe API call forwarded through the proxy with the vault key label, timestamp, endpoint, and response status. For a Databricks billing investigation — for example, a customer reports being charged twice for July — query the audit log for all charges filtered by vault key label (databricks-billing-2026-07-*). If two entries appear for the same customer with the same idempotency key and both returned HTTP 200, Stripe deduplicated them (only one real charge). If two entries appear with different idempotency keys and different vault keys, two separate notebook runs reached the Stripe call independently — the idempotency key was not stable, or two runs used different billing period parameters. Cross-reference the Databricks run IDs from each vault key label with the run history to pinpoint whether the duplicate came from a notebook retry or a concurrent run.