Apache Pulsar Batch Receiver and Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Apache Pulsar’s batchReceive() API is a distinct consumer interface from the standard receive() loop covered in the Pulsar integration guide: it returns a Messages<T> object holding up to maxNumMessages messages, and the acknowledgment semantics are different in ways that introduce Stripe billing failure modes that don’t exist in per-message receive loops. This post covers three failure modes specific to batchReceive() — each arising from the batch receiver’s atomic acknowledgment model, its per-batch-object negative acknowledgment API, and the common pattern of generating a per-batch identifier to “trace” which billing run a charge came from.

The first failure mode: consumer.acknowledge(messages) acknowledges all messages in the Messages<T> object atomically. There is no API to partially acknowledge a batch object. A billing consumer that iterates the batch, calls stripe.charges.create() for each customer, and then calls consumer.acknowledge(messages) at the end of the loop creates a crash window between the last Stripe call and the acknowledge call where any crash (OOM kill, pod eviction, SIGKILL from a rolling deploy) causes the Pulsar broker to redeliver the entire batch — including every customer that was already charged. The broker has no cursor advancement record because no ack was sent for any message in the batch.

The second failure mode: consumer.negativeAcknowledge(messages) — called on the batch object — triggers redelivery of every message in the batch, not just the one that failed. A developer who catches a stripe.error.Timeout on message 23 of 50 and calls consumer.negative_acknowledge(batch) (because the batch object is what’s in scope and the error happened during batch iteration) redelivers all 50 messages after the negativeAckRedeliveryDelay. Messages 1 through 22 were charged successfully. Their customers are billed again 30 seconds later.

The third failure mode: per-batch UUID or batch-start timestamp generated inside the batchReceive() call loop to “trace” which billing run a charge came from, then included in the Stripe idempotency key. When the batch is redelivered on ack timeout or consumer restart, a new batchReceive() call generates a new UUID. The idempotency keys for the redelivered messages differ from the keys used in the original run. Stripe creates ch_B for every customer charged before the crash.

This post covers all three failure modes with Java and Python code, content-hash idempotency keys that exclude all per-batch generation metadata (batch UUID, batch-start timestamp, batch sequence counter), pre-flight PostgreSQL checks with ON CONFLICT DO NOTHING, and per-billing-period vault keys via a spend-cap proxy capped at expected_total × 1.10 — the three-layer governance pattern that closes all three failure modes without changing the BatchReceivePolicy configuration or the Pulsar subscription type.

Failure mode 1: consumer.acknowledge(messages) is atomic — crash between processing message 47 of 50 and the batch ack call redelivers the entire batch, re-billing already-charged customers

The batchReceive() API returns a Messages<T> object when either maxNumMessages messages are available or the BatchReceivePolicy.timeout fires. The common billing pattern: iterate the Messages<T> object, call stripe.charges.create() for each customer, and then call consumer.acknowledge(messages) to advance the subscription cursor for all messages in the batch.

The danger is in the acknowledge call’s atomicity. consumer.acknowledge(messages) sends a single acknowledgment to the Pulsar broker that covers all message IDs in the Messages<T> object. There is no API on the Messages<T> object to acknowledge a subset. The broker treats all messages in the batch as unacknowledged until this one call arrives. If the consumer is killed at any point between the first stripe.charges.create() call and the acknowledge(messages) call — after processing message 1, after processing message 47, or after processing all 50 but before the network send of the ack — the broker redelivers the entire batch.

Note: individual message acknowledgment within a batch-received set is possible by calling consumer.acknowledge(message) on each Message<T> as it is processed. But developers who use the batch form — which is the natural API when the Messages<T> object is what the documentation shows in batch receiver examples — create the all-or-nothing window. The crash window is the duration of the entire billing loop: with 50 customers at ~300ms per Stripe call, the window is up to 15 seconds long.

// Java Pulsar batch receiver — atomic acknowledge (unsafe pattern)
import org.apache.pulsar.client.api.*;

import java.util.concurrent.TimeUnit;

public class BatchBillingConsumer {

    public void run(PulsarClient client) throws PulsarClientException {
        Consumer<UsageEvent> consumer = client.newConsumer(Schema.JSON(UsageEvent.class))
                .topic("persistent://billing/ns/usage-events")
                .subscriptionName("billing-batch-sub")
                .subscriptionType(SubscriptionType.Exclusive)
                .batchReceivePolicy(BatchReceivePolicy.builder()
                        .maxNumMessages(50)
                        .timeout(5, TimeUnit.SECONDS)
                        .build())
                .subscribe();

        StripeClient stripe = new StripeClient(System.getenv("STRIPE_KEY"));

        while (true) {
            Messages<UsageEvent> batch = consumer.batchReceive();
            // batch contains up to 50 messages

            for (Message<UsageEvent> msg : batch) {
                UsageEvent event = msg.getValue();

                // Unsafe: no idempotency key scoped to business fields only.
                // If we use any per-call UUID or any value derived from msg
                // metadata that changes on redelivery, the redelivered batch
                // will generate different keys and Stripe creates ch_B.
                ChargeCreateParams params = ChargeCreateParams.builder()
                        .setAmount(event.amountCents)
                        .setCurrency("usd")
                        .setCustomer(event.stripeCustomerId)
                        .build();

                stripe.charges().create(params);
                // stripe.charges.create() called for message 47 — ch_A created.
                // JVM receives SIGKILL here (rolling deploy, OOM, spot eviction).
                // consumer.acknowledge(batch) never executes.
            }

            // Unsafe: if we are killed between any stripe call above and this line,
            // the broker redelivers the entire 50-message batch on ack timeout.
            // Messages 1-47 were already charged. Stripe creates ch_B through ch_BZ
            // for each one. Pulsar topic stats show: pending_ack_count += 50
            // (all messages return to the PendingAck queue), no error logged.
            consumer.acknowledge(batch);
        }
    }
}

The failure is entirely invisible in Pulsar monitoring. Pulsar topic stats report msgInCounter increasing by 50 on redelivery (as expected for any new batch). subscriptions[].msgBacklog does not increase because the redelivered messages were already consumed — they re-enter the consumer’s dispatcher from the pending-ack store, not from the topic backlog. The subscription cursor does not advance because no ack was sent. From the broker’s perspective, the batch receiver received 50 messages, did not acknowledge them within the ackTimeout window, and they were redelivered. Correct broker behavior. Incorrect billing outcome.

A key distinction from the cumulative ack failure mode covered in the earlier Pulsar guide: the cumulative ack failure mode involves a standard receive() loop where the consumer explicitly configures AckType.Cumulative and sends a single cumulative ack after the loop. The batch receiver failure mode uses the batchReceive() API — the batch Messages<T> object, not ack mode configuration, is what makes acknowledgment all-or-nothing. A developer who migrates from a receive() loop with individual acks to a batchReceive() loop with batch ack changes the acknowledgment semantics without changing ack mode, and the new failure mode arrives silently.

Failure mode 2: consumer.negativeAcknowledge(messages) on a per-message Stripe exception redelivers the entire batch including already-charged customers

The Pulsar consumer API has two negative acknowledgment signatures: consumer.negativeAcknowledge(Message<T> message) which redelivers a single message, and consumer.negativeAcknowledge(Messages<T> messages) which redelivers every message in the batch object. Both are valid API calls. The batch form is the mistake that causes duplicate billing.

The failure pattern: the billing consumer iterates the batch, processing customers one by one. On customer 23 of 50, stripe.charges.create() throws stripe.error.Timeout — the network between the billing service and Stripe timed out after the configured SDK request timeout (typically 30 seconds). The developer catches the exception in the outer loop’s try/except block. The messages batch object is in scope (the outer batchReceive() variable). The developer writes consumer.negative_acknowledge(messages) to signal to Pulsar that the batch failed and should be retried. After negativeAckRedeliveryDelay (default 1 minute, often configured to 30 seconds), Pulsar redelivers all 50 messages. Customers 1 through 22 — charged successfully in the original batch iteration — are billed again.

# Python Pulsar batch receiver — negativeAcknowledge(batch) blast radius (unsafe)
import pulsar
import stripe
import hashlib

client = pulsar.Client("pulsar://localhost:6650")

consumer = client.subscribe(
    "persistent://billing/ns/usage-events",
    "billing-batch-sub",
    consumer_type=pulsar.ConsumerType.Exclusive,
    schema=pulsar.schema.JsonSchema(UsageEvent),
    batch_receive_policy=pulsar.BatchReceivePolicy(
        max_num_messages=50,
        timeout_ms=5000,
    ),
    negative_ack_redelivery_delay_ms=30_000,  # 30 seconds
)

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

while True:
    try:
        batch = consumer.batch_receive()  # returns up to 50 messages
    except Exception:
        continue

    try:
        for msg in batch:
            event = msg.value()

            # No idempotency key — or worse, one derived from msg.message_id()
            # which is stable but does not help when the key for msg 23 differs
            # from what was used on the first attempt (see FM3 below).
            stripe.Charge.create(
                amount=event.amount_cents,
                currency="usd",
                customer=event.stripe_customer_id,
            )
            # Stripe call succeeds for customers 1-22.
            # On customer 23, stripe.error.Timeout is raised.

    except stripe.error.Timeout:
        # Unsafe: calling negative_acknowledge on the entire batch object
        # schedules redelivery of ALL 50 messages — not just message 23.
        # Customers 1-22 were charged successfully. Their billing messages
        # are redelivered 30 seconds later and Stripe creates ch_B for each.
        # Correct call would be: consumer.negative_acknowledge(msg)
        # to redeliver only message 23 (customer 23, the one that timed out).
        consumer.negative_acknowledge(batch)

    else:
        # Only reached if no exception — never reached in the timeout scenario.
        consumer.acknowledge(batch)

What makes this failure mode particularly persistent: the correct fix looks wrong at a glance. consumer.negative_acknowledge(msg) passes the individual Message object from inside the for loop, not the Messages batch object from batch_receive(). But the for loop already advanced past message 23 when the exception was caught in the outer handler. If the exception handler is outside the for loop (as in the example above), msg is the loop variable from the last iteration before the exception, which may or may not be message 23 depending on Python scoping and where in the iteration the exception propagated. The correct pattern is to per-message exception handling inside the for loop, which means restructuring the entire batch processing loop. Many developers take the shortcut of batch-level NAK because it is the only object available in the outer handler, and the duplicates are invisible until a customer calls about a double charge.

A further variant: the developer wraps the entire batch loop in a single try/except and calls consumer.acknowledge(batch) in the success path and consumer.negative_acknowledge(batch) in the exception path — a reasonable transaction-like pattern for database writes, but not for Stripe charges where each message is a separate financial transaction. Unlike a database transaction that is all-or-nothing by design, Stripe charges are individual API calls that cannot be rolled back. Acknowledging message 22 retroactively after acknowledging messages 1 through 21 does not undo the charges for 1 through 21. The billing loop must treat each message independently, not as a unit of work.

Failure mode 3: per-batch UUID or batch-start timestamp in the Stripe idempotency key — new value on every batchReceive() call defeats idempotency on ack-timeout redelivery

This failure mode is a version of a broader class of idempotency key mistakes, but the batchReceive() API creates a specific and common trigger: developers who add per-batch tracing to their billing runs generate a batch_run_id — a UUID or timestamp — at the start of each batchReceive() call and include it in the Stripe idempotency key. The intent is to make each billing run uniquely identifiable in Stripe’s dashboard or in internal audit logs. The problem: the batch_run_id is generated at call time, not at message publication time. When the batch is redelivered — because the consumer crashed before calling acknowledge(messages), or because the ack timeout fired during a long Stripe response sequence — a new batchReceive() call produces a new batch_run_id. Every Stripe idempotency key in the redelivered batch is different from the keys used in the original batch call. Stripe has no cache entry for the new keys and creates fresh charges.

# Python Pulsar batch receiver — per-batch UUID in idempotency key (unsafe)
import pulsar
import stripe
import hashlib
import uuid
import os

client = pulsar.Client("pulsar://localhost:6650")
consumer = client.subscribe(
    "persistent://billing/ns/usage-events",
    "billing-batch-sub",
    consumer_type=pulsar.ConsumerType.Exclusive,
    schema=pulsar.schema.JsonSchema(UsageEvent),
    batch_receive_policy=pulsar.BatchReceivePolicy(
        max_num_messages=50,
        timeout_ms=5000,
    ),
)

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

while True:
    batch = consumer.batch_receive()

    # Per-batch UUID — generated fresh on EVERY batchReceive() call,
    # including ack-timeout-triggered redeliveries of the same messages.
    # Original batch: batch_run_id = "a1b2c3d4-..."
    # Redelivered batch: batch_run_id = "f9e8d7c6-..."  (new UUID)
    batch_run_id = str(uuid.uuid4())

    for msg in batch:
        event = msg.value()

        # Unsafe: key includes batch_run_id which changes per batchReceive() call.
        # Original:   sha256("a1b2c3d4-...:cust_123:Q3-2026") → ch_A
        # Redelivery: sha256("f9e8d7c6-...:cust_123:Q3-2026") → ch_B
        raw = f"{batch_run_id}:{event.customer_id}:{event.billing_period}"
        idempotency_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

        stripe.Charge.create(
            amount=event.amount_cents,
            currency="usd",
            customer=event.stripe_customer_id,
            idempotency_key=idempotency_key,
        )

    # Consumer is killed here before acknowledge() executes.
    # Ack timeout fires. Broker redelivers all 50 messages.
    # Next batchReceive() → new batch_run_id → new keys for all 50 messages.
    # Stripe creates ch_B for every customer charged in the first batch run.
    consumer.acknowledge(batch)

A variant: the developer uses the batchReceive() call’s wall-clock timestamp rather than a UUID — batch_start_ts = int(time.time() * 1000) — as the per-batch component. The same problem: the timestamp changes between the original batchReceive() call and the redelivery call. Original batch at 14:00:00.123, redelivered batch at 14:00:47.892 (after 47 seconds: ack timeout fires, broker waits, consumer restarts, new batchReceive() executes). The timestamp-based key is different. Stripe creates ch_B.

A subtler variant: the developer uses the Pulsar message’s msg.message_id() as the idempotency key seed rather than a batch-level UUID. The message ID is stable across redeliveries — the same message in the same partition has the same message ID on every delivery. This is safe as a key seed if the developer uses only business-identity fields plus the stable message ID. The failure appears when the developer combines the stable message ID with a per-batch value: sha256(msg.message_id() + batch_run_id + customer_id). The msg.message_id() is stable; the batch_run_id changes; the combined key changes. Stripe creates ch_B. The fix is to remove the batch_run_id from the key entirely and use only business-identity fields.

The fix: content-hash idempotency key + pre-flight database check + vault key

All three failure modes share the same root cause: the billing function fires stripe.charges.create() multiple times for the same customer in the same billing period, and the Stripe idempotency key either changes between calls (FM1 when no key is used, FM3 when the key includes per-batch metadata) or the same key is not used exclusively for the failing message (FM2 when the batch NAK redelivers already-keyed messages with correct keys but the billing function would have called them with the same keys anyway — FM2 is most dangerous when no key is used at all). The fix has two independent layers that close all three modes.

1. Content-hash idempotency key stable across all Pulsar batch receiver redeliveries

The Stripe idempotency key must be derived from the billing event’s business-identity fields only — fields present in the message payload that are identical on every possible redelivery of the same billing event. These are customer_id and billing_period (and optionally a stable transaction type label). The key must not include any value generated or captured at the time of the batchReceive() call: no uuid.uuid4(), no time.time(), no batch_run_id, no batch sequence counter, no time.monotonic(), no iteration index within the batch. The key must also not include Pulsar delivery metadata that changes across redeliveries: message redelivery count, consumer name (changes when the consumer restarts with a new instance ID), or subscription cursor position.

# Python Pulsar batch receiver — safe pattern
import pulsar
import stripe
import hashlib
import psycopg2
import os

client = pulsar.Client("pulsar://localhost:6650")
consumer = client.subscribe(
    "persistent://billing/ns/usage-events",
    "billing-batch-sub",
    consumer_type=pulsar.ConsumerType.Exclusive,
    schema=pulsar.schema.JsonSchema(UsageEvent),
    batch_receive_policy=pulsar.BatchReceivePolicy(
        max_num_messages=50,
        timeout_ms=5000,
    ),
    negative_ack_redelivery_delay_ms=30_000,
)

stripe.api_key = os.environ["STRIPE_KEY"]
db = psycopg2.connect(os.environ["DATABASE_URL"])

def make_idempotency_key(customer_id: str, billing_period: str) -> str:
    # Excludes: batch_run_id, batch_start_ts, uuid.uuid4(), time.time(),
    # msg.message_id() (can be included if needed, it is stable, but not required),
    # consumer instance ID, batch sequence counter, iteration index.
    raw = f"{customer_id}:{billing_period}:pulsar-batch-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

while True:
    batch = consumer.batch_receive()
    # No batch_run_id generated here. Nothing per-batchReceive() call.

    for msg in batch:
        event = msg.value()
        idempotency_key = make_idempotency_key(event.customer_id, event.billing_period)

        with db.cursor() as cur:
            # Pre-flight check: write billing intent to PostgreSQL before Stripe call.
            # ON CONFLICT DO NOTHING: if row already exists, rowcount == 0 → skip Stripe.
            # This closes FM1 and FM3 unconditionally (outside Stripe's 24h window too).
            # This closes FM2 by preventing re-billing when the batch is NAKed and
            # the already-charged messages are redelivered.
            cur.execute(
                """
                INSERT INTO billing_records (customer_id, billing_period, idempotency_key)
                VALUES (%s, %s, %s)
                ON CONFLICT (customer_id, billing_period) DO NOTHING
                """,
                (event.customer_id, event.billing_period, idempotency_key),
            )
            rows_affected = cur.rowcount
            db.commit()

        if rows_affected == 0:
            # Pre-flight row already exists — customer was billed in a prior batch run.
            # Acknowledge this individual message and skip Stripe.
            consumer.acknowledge(msg)
            continue

        try:
            charge = stripe.Charge.create(
                amount=event.amount_cents,
                currency="usd",
                customer=event.stripe_customer_id,
                idempotency_key=idempotency_key,
            )

            with db.cursor() as cur:
                cur.execute(
                    "UPDATE billing_records SET charge_id = %s, status = 'BILLED' "
                    "WHERE customer_id = %s AND billing_period = %s",
                    (charge["id"], event.customer_id, event.billing_period),
                )
                db.commit()

            consumer.acknowledge(msg)  # per-message ack — safe within batchReceive() loop

        except stripe.error.CardError:
            # Hard decline — acknowledge to remove from queue, log the decline.
            with db.cursor() as cur:
                cur.execute(
                    "UPDATE billing_records SET status = 'DECLINED' "
                    "WHERE customer_id = %s AND billing_period = %s",
                    (event.customer_id, event.billing_period),
                )
                db.commit()
            consumer.acknowledge(msg)

        except stripe.error.Timeout:
            # Network timeout — charge may or may not have been created server-side.
            # Content-hash key handles redelivery: Stripe returns ch_A if it was created.
            # Pre-flight row exists, so retry won't create ch_B even outside 24h window.
            # NAK only this message — not the entire batch.
            consumer.negative_acknowledge(msg)

        except Exception:
            consumer.negative_acknowledge(msg)

The key structural changes: (1) no per-batch UUID or timestamp is generated — the make_idempotency_key() function takes only business-identity fields from the message payload; (2) the acknowledge calls are per-message (consumer.acknowledge(msg) and consumer.negative_acknowledge(msg)), not per-batch (consumer.acknowledge(batch) and consumer.negative_acknowledge(batch)); (3) the pre-flight database INSERT executes before each Stripe call individually, not once for the whole batch.

2. Pre-flight database check with ON CONFLICT DO NOTHING

The content-hash idempotency key closes FM1 and FM3 within Stripe’s 24-hour idempotency window: the same key on a redelivered message returns ch_A silently. FM2 is closed within 24 hours for messages processed before the NAK call. The pre-flight PostgreSQL check closes all three modes unconditionally, regardless of how much time has elapsed since the original charge.

The pre-flight check: INSERT INTO billing_records (customer_id, billing_period, idempotency_key) VALUES (%s, %s, %s) ON CONFLICT (customer_id, billing_period) DO NOTHING. If rowcount == 0, the customer was already billed. Skip the Stripe call and acknowledge the message. No second charge, no retry. The PostgreSQL UNIQUE (customer_id, billing_period) constraint is the durable deduplication layer that survives consumer restarts, broker redeliveries, ack timeouts, and NAK-triggered redeliveries of the entire batch.

Write-before-call ordering: the INSERT must commit to PostgreSQL before stripe.charges.create() is called. If the billing process is killed between the PostgreSQL commit and the Stripe call, the customer is not charged (the pre-flight row blocks a later re-delivery from charging). If killed between the Stripe call completing and UPDATE billing_records SET charge_id, the pre-flight row blocks the next delivery (rowcount == 0). The charge_id column can be backfilled by a reconciliation job that queries Stripe using the stable idempotency key.

3. Per-billing-period vault key capped at expected_total × 1.10

A Keybrake vault key issued per billing period with a spend cap of expected_total × 1.10 provides a hard backstop at the Stripe API boundary when both the content-hash key and the pre-flight check are transiently unable to prevent a duplicate charge. This occurs when the pre-flight PostgreSQL database is unreachable (connection pool exhaustion, network partition between the Pulsar consumer pod and the database) at the moment of a batch redelivery. Without the vault cap, the Stripe call proceeds with the content-hash key (which returns ch_A if within 24 hours, or creates ch_B if outside 24 hours). With the vault cap, the proxy compares the total charges for this billing period against the cap and rejects any charge that would exceed it, regardless of the idempotency key value, providing defense in depth when both upstream deduplication layers fail simultaneously.

Gap analysis: Pulsar batch receiver-specific scenarios

Batch receiver with BatchReceivePolicy.maxNumBytes — partial batch on byte limit with early ack timeout

The BatchReceivePolicy can be configured with a maxNumBytes limit in addition to or instead of maxNumMessages. When billing events are variable-size (e.g., usage events with large metadata payloads), a batch may return fewer than maxNumMessages messages because the byte limit was reached first. If the consumer processes this partial batch, calls Stripe for each customer, and then is killed before the acknowledge(messages) call, the broker redelivers the partial batch. The redelivery count for each message in the partial batch increments (visible via msg.redelivery_count() in Python, msg.getRedeliveryCount() in Java), but the content-hash idempotency key is not affected by the redelivery count — the same key is used on the retry. Within 24 hours, Stripe returns ch_A. Beyond 24 hours, the pre-flight PostgreSQL check prevents ch_B.

Batch receiver ack timeout vs. negativeAckRedeliveryDelay — two independent redelivery timers

The Pulsar consumer has two independent redelivery mechanisms that interact with batch receivers: the ackTimeout (time after message delivery before the broker redelivers unacknowledged messages) and the negativeAckRedeliveryDelay (time after negativeAcknowledge() before redelivery). When a batch billing consumer is slow — 50 messages at 300ms each Stripe call = 15 seconds of processing — the ackTimeout (if configured to 10 seconds) can fire mid-batch, redelivering the unacknowledged messages from the first batch to a second consumer. This is a variant of FM1 where the redelivery is triggered not by a consumer crash but by processing time exceeding the ack timeout. The safe pattern: either set ackTimeout to significantly longer than the maximum expected batch processing time, or use per-message acknowledgment within the batch loop (which fires acks continuously as messages are processed, rather than holding all acks until the end). The content-hash key and pre-flight check close the duplicate charge regardless of which timer triggers the redelivery.

Batch receiver in multi-topic subscription — same billing event delivered from two topics in the same batch

Pulsar supports multi-topic subscriptions where a single consumer subscribes to a list of topics. The batchReceive() call returns messages from any of the subscribed topics interleaved in a single Messages<T> object. In geo-replicated Pulsar deployments where the same billing event is published to a source topic and replicated to a target cluster topic, a billing consumer subscribed to both the source and the replica topic can receive two copies of the same billing event in the same batch — one from each topic. If the idempotency key is derived from only customer_id and billing_period (the content-hash approach), both copies produce the same key and the second call returns ch_A from Stripe’s idempotency cache. If the key includes the topic name (e.g., sha256(msg.topic_name() + ":" + customer_id + ":" + billing_period)), the two copies produce different keys (different topic names) and Stripe creates ch_B. The fix: never include msg.topic_name(), msg.partition(), or any Pulsar routing metadata in the idempotency key.

Frequently asked questions

Can I use consumer.acknowledge(msg) for each message inside the batchReceive() loop to avoid the atomic acknowledge problem?

Yes, and this is the recommended safe pattern. The Messages<T> object returned by batchReceive() is iterable; each element is a Message<T> on which you can call consumer.acknowledge(message) individually as it is processed. This means each message is acknowledged as soon as its Stripe call completes and the pre-flight database is updated. A crash mid-batch redelivers only the unacknowledged messages (those not yet processed and individually acked), not the entire batch. The batch receive performance benefit is still achieved — the broker delivers 50 messages in one network round-trip — and the per-message ack means the crash window is scoped to each individual message rather than the entire batch processing window. The same approach applies to negative acknowledgment: call consumer.negative_acknowledge(msg) on the individual failing message, not consumer.negative_acknowledge(messages) on the batch object.

Does the Pulsar broker track which messages in a batch were processed before a crash?

No. The Pulsar broker’s view of a batch-received set is binary: either all messages in the set have been acknowledged (cursor advances past the batch), or none have (redelivery on ack timeout). The broker does not receive or store partial acknowledgment state for a Messages<T> batch that was delivered via batchReceive(). The per-message individual acknowledge calls described above are the mechanism for partial advancement: each consumer.acknowledge(msg) call advances the subscription cursor for that specific message ID, allowing the broker to track which messages within a batch’s sequence range have been acknowledged and which have not. This is different from cumulative acknowledgment (which advances the cursor to a specific sequence number, implicitly acknowledging all prior messages) and from batch object acknowledgment (which acknowledges all messages in the batch atomically).

If I include the Pulsar msg.message_id() in the Stripe idempotency key, is that safe?

A stable message ID is safe as an idempotency key seed, but only if it is the only non-business-field component in the key. msg.message_id() in Pulsar includes the ledger ID, entry ID, and partition index — values assigned by the broker at message publication time and stable across redeliveries of the same message. So sha256(str(msg.message_id()) + ":" + customer_id + ":" + billing_period) is stable across redeliveries and will not produce FM3. The problem appears when the developer adds a per-batch value alongside the message ID: sha256(batch_run_id + ":" + str(msg.message_id()) + ":" + customer_id). The msg.message_id() is stable; the batch_run_id is not. The combination is unstable. If you use message ID, use it alone plus business fields — never combined with any value generated at batchReceive() call time. The content-hash approach (business fields only, no message ID) is simpler and eliminates any risk of accidentally introducing a per-invocation component.

Does the Keybrake vault key cap need to cover all 50 messages in a batch or just one customer?

The vault key is issued per customer per billing period, not per batch. Each customer in the batch has their own vault key with a spend cap of expected_total × 1.10. The proxy evaluates the cap against the total charges on that customer’s vault key for the current billing period. A 50-message batch creates 50 independent vault keys (one per customer). If customers 1 through 22 were charged in the first batch run and the batch is redelivered, each of those 22 customers’ vault keys already shows expected_total × 1.0 in charges. A second charge attempt for any of them would push their total above the cap and is rejected by the proxy before reaching Stripe. Customers 23 through 50 were not charged in the first batch run; their vault key charges are zero and the second delivery is allowed through (as it should be, since it is their first charge). The cap is fine-grained to the customer-billing-period scope, not the batch scope.