Azure Event Hubs and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Azure Event Hubs is a partitioned, append-only event log. Processors read events by offset, checkpoint progress to Azure Blob Storage, and hold partition ownership via blob leases. When Event Hubs drives billing, three Event Hubs-specific behaviors introduce Stripe double-charge failure modes that Event Hubs metrics — IncomingMessages, OutgoingMessages, ThrottledRequests — do not surface as billing risk.

This post covers those three failure modes with azure-eventhub Python code, content-hash idempotency keys stable across partition re-assignments, checkpoint recoveries, and producer retries, per-billing-period vault keys via a spend-cap proxy, and pre-flight database checks — the two-layer governance approach that closes all three without changing Event Hubs configuration, partition count, or consumer group settings.

Failure mode 1: Blob Storage checkpoint write fails after stripe.charges.create() returns — the next partition owner re-processes from the uncommitted offset

Every Azure Event Hubs processor that uses EventProcessorClient tracks its position in each partition by writing a checkpoint to Azure Blob Storage. The checkpoint records the partition ID, offset, and sequence number of the last successfully processed event. When a processor reads event E, processes it, and calls partition_context.update_checkpoint(event), the checkpoint blob for that partition is updated to offset E+1. The next time any processor takes ownership of that partition — whether because the current processor crashed, its lease expired, or a deployment rollout replaced it — the new processor reads from the checkpoint position and receives event E+1. If the checkpoint was never written, the new processor reads from the last committed checkpoint, which may be event E-1 or earlier, and re-processes event E.

The failure scenario: a billing processor reads event E from partition 3. The event body contains customer_id, billing_period, and amount_cents. The processor calls stripe.charges.create() — ch_A is created on Stripe’s servers, returning after 18 seconds. The processor now calls partition_context.update_checkpoint(event). This write goes to Azure Blob Storage. If Blob Storage is under load — the storage account hosting the checkpoint container is throttled because concurrent checkpoint writes from all partitions (Event Hubs can have 32 partitions by default) saturate the storage account’s IOPS limit — the checkpoint write returns HttpResponseError: 503 Server Busy. Alternatively, the processor receives an OOM kill signal from the Kubernetes scheduler between the Stripe call returning and the update_checkpoint() call executing — the checkpoint write is never issued. In both cases, the Blob Storage checkpoint for partition 3 still points to E-1.

When the EventProcessorClient’s load balancer detects that partition 3 is unowned (lease expired or processor died), it assigns partition 3 to another processor instance. That instance reads from the last committed checkpoint — event E — processes event E, and calls stripe.charges.create() with no idempotency key. Stripe receives a fresh charges.create request for the same customer and creates ch_B. The second processor’s update_checkpoint() succeeds. From Event Hubs’ perspective, partition 3 had its checkpoint updated once. From Stripe’s perspective, the customer was charged twice.

import asyncio
import json
import os
import stripe
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

CONNECTION_STR = os.environ["EVENTHUB_CONNECTION_STR"]
EVENTHUB_NAME = os.environ["EVENTHUB_NAME"]
STORAGE_CONNECTION_STR = os.environ["STORAGE_CONNECTION_STR"]
CHECKPOINT_CONTAINER = "billing-checkpoints"

async def on_event(partition_context, event):
    body = json.loads(event.body_as_str())
    customer_id = body["customer_id"]
    billing_period = body["billing_period"]

    # stripe.charges.create() can take 18+ seconds under load.
    # ch_A is created on Stripe's servers when this call returns.
    charge = await asyncio.to_thread(
        stripe.Charge.create,
        amount=body["amount_cents"],
        currency="usd",
        customer=body["stripe_customer_id"],
        description=f"Usage billing for {billing_period}",
        # no idempotency_key — re-processing creates ch_B
        api_key=os.environ["STRIPE_SECRET_KEY"],
    )

    # update_checkpoint writes to Azure Blob Storage.
    # If this write fails (503 throttle, network timeout, OOM kill between
    # Stripe returning and this line executing), the checkpoint stays at E-1.
    # The next processor to own partition 3 re-reads event E and calls
    # stripe.charges.create() again — creating ch_B alongside ch_A.
    await partition_context.update_checkpoint(event)
    print(f"Charged {customer_id}: {charge['id']}")


async def main():
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        STORAGE_CONNECTION_STR, CHECKPOINT_CONTAINER
    )
    client = EventHubConsumerClient.from_connection_string(
        CONNECTION_STR,
        consumer_group="$Default",
        eventhub_name=EVENTHUB_NAME,
        checkpoint_store=checkpoint_store,
    )
    async with client:
        await client.receive(on_event=on_event, starting_position="-1")

The checkpoint-failure mode is particularly insidious because it has two distinct triggers that occur independently. The storage throttling trigger is correlated with billing load: a billing run that processes thousands of events per second generates thousands of concurrent checkpoint writes, which compete for Blob Storage IOPS and are most likely to fail precisely when the Event Hubs throughput is highest. The OOM kill trigger is correlated with memory pressure: billing callbacks that hold event body data, Stripe response objects, and database connection state can consume significant heap, and Kubernetes OOM eviction can fire at the exact moment between a successful Stripe call and an incomplete checkpoint. Both failure modes produce the same outcome: a committed Stripe charge and an uncommitted checkpoint, with no indication in Event Hubs logs or Azure Monitor metrics that the event will be re-processed.

Failure mode 2: Blob Storage lease renewal fails under storage throttling — a competing EventProcessorClient claims the partition while stripe.charges.create() is in-flight

EventProcessorClient manages partition ownership through Azure Blob Storage leases. Each partition is represented by a blob in the checkpoint container, and the processor that holds the active lease on that blob owns the partition. The lease duration defaults to 30 seconds. A background task within each processor instance renews its leases on a schedule — typically every 10–15 seconds — by calling BlobLeaseClient.renew(). If the background renewal call fails, the lease expires at the end of its 30-second duration. Any other EventProcessorClient instance that checks the blob during the load-balancing cycle and finds the lease expired can acquire ownership of the partition.

The critical difference from other messaging systems: Azure Event Hubs’ partition ownership has no equivalent to SQS’s change_message_visibility or Azure Service Bus’s AutoLockRenewer at the event level. Unlike Azure Service Bus, where each message has its own lock token that can be renewed independently, Event Hubs partition ownership is coarser — it applies to the entire partition, not to individual events. A billing callback processing event E on partition 3 cannot extend the partition’s lease independently from the background renewal thread. If the renewal thread fails, the entire partition can be stolen by a competing instance while event E is still being processed.

The failure scenario: processor A owns partition 3. It reads event E and calls stripe.charges.create() at T=0. The call takes 20 seconds. At T=10, the background lease renewal thread calls BlobLeaseClient.renew() for partition 3. Blob Storage returns HttpResponseError: 429 TooManyRequests — the storage account is throttled from the volume of concurrent checkpoint writes and lease renewals across all partitions. The renewal thread logs the error and will retry at T=25. The existing lease expires at T=30 (issued at T=0 with 30-second duration). At T=25, a second processor instance B running the load-balancing loop checks the partition 3 blob and finds the lease has not been renewed recently (remaining lease time is 5 seconds). Instance B acquires the partition 3 lease at T=25, reads from the last committed checkpoint (which points to event E, since processor A’s checkpoint write for event E has not happened yet), and begins processing event E at T=26. Instance B calls stripe.charges.create() for the same customer at T=26. Processor A’s Stripe call returns ch_A at T=20, and processor A attempts partition_context.update_checkpoint(event). The checkpoint write fails with ResourceModifiedError — the blob lease is now held by instance B, not A. Many SDK implementations swallow this error or log it as a warning. Instance B’s Stripe call creates ch_B at T=38 and its checkpoint write succeeds.

import asyncio
import json
import os
import stripe
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

async def on_event(partition_context, event):
    body = json.loads(event.body_as_str())
    customer_id = body["customer_id"]
    billing_period = body["billing_period"]

    # The EventProcessorClient's background lease renewal thread runs
    # on a separate schedule. If Blob Storage is throttled (429 TooManyRequests)
    # and the renewal thread fails, the partition lease expires in 30 seconds.
    # A competing EventProcessorClient instance can claim partition ownership
    # mid-processing — while this Stripe call is in-flight.
    # The competing instance reads from the last committed checkpoint (event E)
    # and calls stripe.charges.create() for the same customer.
    charge = await asyncio.to_thread(
        stripe.Charge.create,
        amount=body["amount_cents"],
        currency="usd",
        customer=body["stripe_customer_id"],
        description=f"Usage billing for {billing_period}",
        # no idempotency_key — competing instance creates ch_B
        api_key=os.environ["STRIPE_SECRET_KEY"],
    )

    # If partition ownership was stolen while we were in stripe.charges.create(),
    # this update_checkpoint() call fails with ResourceModifiedError.
    # Many default error handlers log this as a warning and continue.
    # Instance B's checkpoint call succeeds — billing_period is now at E+1
    # for the instance that created ch_B. Both charges are on the customer's account.
    await partition_context.update_checkpoint(event)
    print(f"Charged {customer_id}: {charge['id']}")

The lease-theft failure is correlated with storage throttling — precisely the condition that is most likely during a high-volume billing run. A billing run that generates 10,000 events across 32 partitions creates 10,000 checkpoint writes and 32 lease renewals every 10 seconds. All of these operations compete for the same Blob Storage account’s throughput limits. The storage throttling that causes lease renewal failure is the same throttling that makes the billing run the most likely period for partition ownership to be contested. Azure Monitor metrics (ThrottledRequests on the storage account) can confirm that throttling occurred, but they do not indicate which specific lease renewal failed or whether a billing event was processed by two concurrent partition owners.

Failure mode 3: EventHubProducerClient retries on network timeout publish duplicate events with different sequence numbers — sequence_number-based idempotency keys treat them as separate billing requests

Unlike SQS, Azure Service Bus, Google Pub/Sub, and Redis Streams — where re-delivery of the same message preserves a stable identity (SQS MessageId, Service Bus message_id within a session, Pub/Sub ack_id for ordered subscriptions) — Azure Event Hubs assigns metadata at ingestion time. The sequence_number, offset, and enqueued_time in event.system_properties are set by Event Hubs when the event is committed to the partition log. They reflect when and where the event entered the log, not the business identity of the event itself.

This becomes a billing failure mode because EventHubProducerClient retries publish calls on transient network errors. A producer sends a billing event for customer C, billing period P, amount $50. The Azure Event Hubs server accepts the event, commits it to partition 5 at offset 10082 with sequence_number=1042, and sends the acknowledgment back to the producer. The acknowledgment is lost in transit — the network connection drops after the server sent the ACK but before the producer’s TCP stack received it. The producer’s SDK times out waiting for the acknowledgment and retries. The retry publishes a second copy of the same billing event to partition 5 (or a different partition, depending on partition key routing). Event Hubs commits the second copy at offset 10083 with sequence_number=1043. Both copies have identical body (same customer_id, billing_period, amount_cents) but different system properties.

A processor that derives its Stripe idempotency key from Event Hubs system properties — sequence_number, offset, or enqueued_time — produces different idempotency keys for the two event copies. When the processor reads offset 10082, it computes key K1 and calls stripe.charges.create(idempotency_key=K1) — ch_A is created. When it reads offset 10083, it computes key K2 and calls stripe.charges.create(idempotency_key=K2) — Stripe treats this as a new request (K2 has never been seen) and creates ch_B. The customer receives two Stripe charges for the same billing period.

import asyncio
import json
import os
import hashlib
import stripe
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

async def on_event_unsafe(partition_context, event):
    body = json.loads(event.body_as_str())
    customer_id = body["customer_id"]
    billing_period = body["billing_period"]

    # UNSAFE: idempotency key derived from Event Hubs system properties.
    # EventHubProducerClient retries on transient network errors.
    # The retry publishes a duplicate event to the same (or different) partition.
    # Event Hubs assigns new sequence_number, offset, and enqueued_time to each copy.
    # This processor computes a different idempotency key for each copy:
    # - offset 10082, sequence_number=1042 → key = sha256("C:P:1042")[:32]
    # - offset 10083, sequence_number=1043 → key = sha256("C:P:1043")[:32]
    # Stripe treats K1 and K2 as separate billing requests and creates ch_A and ch_B.
    seq = event.sequence_number  # changes between producer retries
    bad_key = hashlib.sha256(
        f"{customer_id}:{billing_period}:{seq}".encode()
    ).hexdigest()[:32]

    charge = await asyncio.to_thread(
        stripe.Charge.create,
        amount=body["amount_cents"],
        currency="usd",
        customer=body["stripe_customer_id"],
        description=f"Usage billing for {billing_period}",
        idempotency_key=bad_key,  # different key for each producer retry copy
        api_key=os.environ["STRIPE_SECRET_KEY"],
    )
    await partition_context.update_checkpoint(event)
    print(f"Charged {customer_id}: {charge['id']}")

The producer retry failure mode is subtle because it operates at a different layer than the consumer-side failure modes. The consumer code is not wrong — it processes each event it receives exactly once. The billing duplication occurs because the event stream itself contains two copies of the same billing intent, and a consumer with no business-level deduplication treats them as two legitimate events. Event Hubs’ built-in duplicate detection (duplicate_detection_history_time_window in Standard and Premium tier) is based on user-provided message_id headers set at publish time — it does not apply unless the producer explicitly sets message_id on each EventData. Most producers using EventHubProducerClient.send_batch() do not set message_id, and Event Hubs assigns no stable identity to events at ingestion.

The failure is also possible in fanout architectures: if the billing event stream feeds multiple downstream Event Hubs (via routing rules or custom forwarding), and the forwarding logic retries on timeout, the same billing event can appear in the downstream hub twice with different system properties, defeating any idempotency key derived from those properties.

The fix: content-hash idempotency key + vault key + pre-flight check + write-before-checkpoint

All three failure modes share the same root cause: stripe.charges.create() is called more than once for the same (customer_id, billing_period) pair, and neither Event Hubs’ checkpoint mechanics nor Stripe’s API layer prevents the second call from creating a new charge when the idempotency key is absent, derived from unstable system properties, or rotated. The fix has three independent components that together close all three modes without changing Event Hubs configuration, partition count, or checkpoint frequency.

Layer 1: content-hash idempotency key stable across partition re-assignments and producer retries

The Stripe idempotency key must be derived from the billing event’s stable business fields — fields present in the event body that represent the same billing intent regardless of which partition, offset, or sequence number Event Hubs assigned to the event. The key must not include any Event Hubs system properties that change between producer retry copies or across partition re-assignments. Fields to exclude: sequence_number (assigned at ingestion per partition — a producer retry copy gets a different sequence number), offset (byte offset within the partition log — changes for every event, including retry copies), enqueued_time from event.system_properties (set at ingestion time — the retry copy has a later enqueued_time), and partition_id (the partition number — a producer using round-robin or hash routing may publish the retry copy to a different partition). These are all log-position or ingestion-time properties. They identify where the event sits in the Event Hubs log, not what billing action it represents.

import hashlib
import asyncio
import json
import os
import stripe
import psycopg2
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Derived from stable business identity: customer + billing period only.
    # Must NOT include: sequence_number (assigned at ingestion; different for
    # each producer retry copy), offset (byte position in partition log; different
    # for each event including retry copies), enqueued_time from SystemProperties
    # (set at ingestion time; later for retry copies), or partition_id (changes
    # if the producer routes the retry copy to a different partition).
    # All of these change across producer retry copies and partition re-assignments.
    raw = f"{customer_id}:{billing_period}:azure-event-hubs-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]


def process_billing_event_sync(conn, body: dict) -> dict:
    customer_id = body["customer_id"]
    stripe_customer_id = body["stripe_customer_id"]
    amount_cents = body["amount_cents"]
    billing_period = body["billing_period"]

    idempotency_key = make_idempotency_key(customer_id, billing_period)

    # Pre-flight check: was this customer already billed for this period?
    # Authoritative source: billing_records (PostgreSQL — durable through
    # Event Hubs partition re-assignments and Blob Storage checkpoint failures).
    # This closes all three Event Hubs failure modes:
    # (1) Checkpoint failure: if the first processor wrote billing_records before
    #     update_checkpoint() failed, the re-processing processor's pre-flight
    #     check finds the existing row and skips the Stripe call.
    # (2) Lease theft: if the first processor wrote billing_records before the
    #     partition was stolen, the competing processor's pre-flight check finds
    #     the existing row and skips stripe.charges.create().
    # (3) Producer retry duplicate: if the first event copy was processed and
    #     billing_records was written, the second event copy's processor finds
    #     the existing row and skips — regardless of the different sequence_number.
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT charge_id, status FROM billing_records
            WHERE customer_id = %s AND billing_period = %s
            """,
            (customer_id, billing_period),
        )
        existing = cur.fetchone()

    if existing:
        return {
            "skipped": True,
            "reason": "already_billed",
            "charge_id": existing[0],
        }

    vault_key = os.environ["KEYBRAKE_VAULT_KEY"]

    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=stripe_customer_id,
        description=f"Usage billing for {billing_period}",
        idempotency_key=idempotency_key,
        api_key=vault_key,
    )

    # Write to billing_records BEFORE calling update_checkpoint().
    # Write-before-checkpoint: if update_checkpoint() fails (Blob Storage 503,
    # ResourceModifiedError from lease theft), billing_records already has the row.
    # The re-processing processor's pre-flight check finds it and skips.
    # ON CONFLICT DO NOTHING closes the concurrent lease-theft race:
    # if two processors simultaneously passed the pre-flight check (lease theft
    # fired just after both read no existing row), the unique constraint ensures
    # only one INSERT succeeds — the second is silently ignored.
    # The idempotency key ensures both processors receive ch_A from Stripe.
    with conn.cursor() as cur:
        cur.execute(
            """
            INSERT INTO billing_records
                (customer_id, billing_period, charge_id, amount_cents, status, created_at)
            VALUES (%s, %s, %s, %s, 'charged', NOW())
            ON CONFLICT (customer_id, billing_period) DO NOTHING
            """,
            (customer_id, billing_period, charge["id"], amount_cents),
        )
    conn.commit()

    return {"charge_id": charge["id"], "status": "charged"}

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

The vault key is a per-billing-period Stripe key issued through a spend-cap proxy. The vault key enforces a maximum spend for the billing run, set at 110% of the expected total. For the producer-retry failure mode — where a second event copy is processed after Stripe’s 24-hour idempotency window has closed (a duplicate event held in a partition that was not processed for over a day due to a consumer group lag incident) — the vault key caps worst-case spend at expected_total × 1.10, providing a hard backstop even when the content-hash idempotency key’s 24-hour deduplication window has elapsed.

import requests
import os
import datetime

def create_billing_vault_key(
    billing_period: str,
    expected_total_cents: int,
) -> str:
    resp = requests.post(
        "https://proxy.keybrake.com/vault/keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "vendor": "stripe",
            "label": f"event-hubs-billing-run-{billing_period}",
            "daily_usd_cap": round((expected_total_cents * 1.10) / 100, 2),
            "allowed_endpoints": ["/v1/charges", "/v1/payment_intents"],
            "expires_at": (
                datetime.datetime.utcnow() + datetime.timedelta(hours=6)
            ).isoformat() + "Z",
        },
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

# For a billing run with 500 customers at $25 each = $12,500 expected total:
# - Vault key cap: $13,750 (110% of expected total)
# - Lease theft race: two processors both process the same partition event
#   simultaneously (one processor per event — limited blast radius)
# - Idempotency key deduplicates charges within 24h of original attempt
# - Pre-flight check skips charges where billing_records was already written
# - Vault cap blocks charges past $13,750 (covers expired-idempotency edge case)

Safe Event Hubs billing processor with write-before-checkpoint

import asyncio
import json
import os
import logging
import stripe
import psycopg2
from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore

logger = logging.getLogger(__name__)

conn = psycopg2.connect(os.environ["DATABASE_URL"])


async def on_event(partition_context, event):
    try:
        body = json.loads(event.body_as_str())
        customer_id = body["customer_id"]
        billing_period = body["billing_period"]

        result = await asyncio.to_thread(process_billing_event_sync, conn, body)

        if result.get("skipped"):
            # Pre-flight check found existing billing record.
            # Safe to checkpoint — no Stripe call was made.
            await partition_context.update_checkpoint(event)
            logger.info(
                "Skipped %s/%s: already_billed (%s)",
                customer_id,
                billing_period,
                result.get("charge_id"),
            )
            return

        # billing_records write completed inside process_billing_event_sync.
        # update_checkpoint() AFTER the database write (write-before-checkpoint).
        # If update_checkpoint() fails (Blob Storage 503, ResourceModifiedError
        # from lease theft between Stripe returning and this call executing):
        # - billing_records already has the charge row (durable in PostgreSQL)
        # - the next processor to own this partition re-processes event E
        # - the pre-flight check finds the existing row and skips stripe.charges.create()
        # - the re-processing processor successfully checkpoints past E
        await partition_context.update_checkpoint(event)
        logger.info(
            "Charged %s/%s: %s",
            customer_id,
            billing_period,
            result.get("charge_id"),
        )

    except stripe.error.Timeout:
        # Stripe timed out. ch_A may already exist on Stripe's servers.
        # Do NOT checkpoint — keep event at current offset for re-processing.
        # On re-processing (partition re-assignment or consumer group restart):
        # - content-hash idempotency key returns ch_A within 24h
        # - pre-flight check skips if billing_records was written before timeout
        logger.warning(
            "Stripe timeout for %s/%s — not checkpointing; event will be re-processed",
            body.get("customer_id"),
            body.get("billing_period"),
        )
        # Do not call update_checkpoint(). The event stays at the current offset.
        # EventProcessorClient will re-deliver it after partition re-assignment.

    except stripe.error.CardError as e:
        # Card declined — permanent failure. Write to billing_records
        # with status='card_declined', then checkpoint.
        # Do NOT leave uncheckpointed: the event will be re-processed indefinitely,
        # creating repeated Stripe card-decline events visible to the customer.
        try:
            with conn.cursor() as cur:
                cur.execute(
                    """
                    INSERT INTO billing_records
                        (customer_id, billing_period, charge_id,
                         amount_cents, status, error, created_at)
                    VALUES (%s, %s, NULL, %s, 'card_declined', %s, NOW())
                    ON CONFLICT (customer_id, billing_period) DO NOTHING
                    """,
                    (
                        body["customer_id"],
                        body["billing_period"],
                        body["amount_cents"],
                        str(e),
                    ),
                )
                conn.commit()
        finally:
            await partition_context.update_checkpoint(event)

    except Exception as e:
        logger.error(
            "Billing failed for %s: %s — not checkpointing; event will be re-processed",
            body.get("customer_id"),
            e,
        )
        # For unknown errors, do not checkpoint. The idempotency key and
        # pre-flight check make the re-processed delivery safe regardless
        # of what the current delivery did.


async def main():
    checkpoint_store = BlobCheckpointStore.from_connection_string(
        os.environ["STORAGE_CONNECTION_STR"], "billing-checkpoints"
    )
    client = EventHubConsumerClient.from_connection_string(
        os.environ["EVENTHUB_CONNECTION_STR"],
        consumer_group="$Default",
        eventhub_name=os.environ["EVENTHUB_NAME"],
        checkpoint_store=checkpoint_store,
    )
    async with client:
        await client.receive(on_event=on_event, starting_position="-1")


if __name__ == "__main__":
    asyncio.run(main())

Comparison: protection levels across all three failure modes

FAQ

Does enabling Event Hubs duplicate detection (duplicate_detection_history_time_window) prevent producer retry duplicates from reaching billing consumers?

Duplicate detection in Azure Event Hubs Standard and Premium tier works when the producer explicitly sets a message_id property on each EventData before publishing. Event Hubs stores the set of message_id values seen within the configured time window (up to 7 days) and discards incoming events whose message_id has been seen in that window. If the producer sets a stable message_id derived from business fields (e.g., a hash of customer_id + billing_period), duplicate detection filters producer retry copies before they enter the partition log — eliminating failure mode 3 at the infrastructure layer. However, most billing systems using EventHubProducerClient.send_batch() do not set message_id, and Event Hubs assigns no default message identity. Duplicate detection also requires Standard or Premium tier — it is not available on Basic tier. The content-hash idempotency key and pre-flight database check close the producer-retry failure mode regardless of tier, message_id configuration, or whether duplicate detection is enabled, because they operate at the consumer level on stable business fields from the event body.

Can Event Hubs consumer group position (earliest vs. latest starting position) prevent re-processing billing events?

The starting position passed to EventHubConsumerClient.receive(starting_position=...) or EventProcessorClient determines where a consumer group begins reading when no committed checkpoint exists. Once a checkpoint is committed for a partition, that position takes precedence over the starting position configuration. The starting position does not prevent re-processing on checkpoint failures — if the checkpoint was committed at offset E-1 (the event before the failed billing event), the consumer will re-read from E regardless of whether the starting position is configured as “earliest” or “latest”. Starting position is relevant only for the very first time a consumer group reads from a partition. It does not provide idempotency or prevent the billing failure modes described here.

Does Event Hubs’ event retention period create a risk that billing events will be re-processed indefinitely?

Azure Event Hubs retains events for a configurable retention period (default 1 day on Standard tier, up to 90 days on Premium tier). A consumer group that falls behind — due to a processing incident, deployment rollout, or consumer group lag — and then resumes processing will re-process events from its last checkpoint, which may be hours or days old. Events processed more than 24 hours after they were originally billed will not be deduplicated by Stripe’s idempotency key (Stripe’s deduplication window is 24 hours). For these re-processed events, the pre-flight database check against billing_records is the primary defense: if the event was originally processed and billing_records was written, the re-processing consumer finds the row and skips the Stripe call regardless of how old the event is. The vault key spend cap provides a secondary backstop for edge cases where billing_records was not written (e.g., the original processor crashed between stripe.charges.create() returning and the database write committing) and the Stripe idempotency window has closed.

How does the write-before-checkpoint pattern interact with Event Hubs’ at-least-once delivery guarantee?

Event Hubs’ at-least-once delivery guarantee means a committed checkpoint is the only mechanism that prevents re-processing. An event is re-processed if the partition is re-assigned to a new processor before its checkpoint is committed. The write-before-checkpoint pattern ensures that the database write to billing_records always precedes update_checkpoint(). If update_checkpoint() fails, the billing_records row is already in PostgreSQL — durable because PostgreSQL WAL is fsynced on commit, unlike Blob Storage checkpoint writes that can fail with 503 throttling errors. The next processor to own the partition re-processes from the uncommitted offset, runs the pre-flight check, finds the billing_records row, and skips the Stripe call safely. This pattern aligns with Event Hubs’ at-least-once guarantee: events may be processed more than once, but the pre-flight check ensures that only one Stripe charge is created per billing period per customer across all deliveries.

Should I use Event Hubs Capture to archive events and recover from consumer group failures?

Event Hubs Capture writes incoming events to Azure Blob Storage or Azure Data Lake Storage Gen2 in Avro format on a configurable time or size trigger (minimum 1 minute or 10 MB). Capture provides a durable archive of all events that can be replayed if the consumer group falls behind or if the retention period for the Event Hub itself is insufficient. However, Capture is an archive mechanism, not an idempotency mechanism. Events replayed from Capture archives will have the same sequence_number and offset as their originals (they are the same events), but will be re-delivered to the consumer as if they were being processed for the first time. The content-hash idempotency key, pre-flight database check, and vault key spend cap protect billing correctness during Capture replays just as they do for normal re-processing — the pre-flight check finds existing billing_records rows and skips Stripe calls for events that were billed during the original processing run.