Keybrake

Spring Cloud Stream Pulsar Stripe Integration: How SCS Pulsar Binder subscriptionType=Shared Plus ackTimeout Redelivers the In-Flight Billing Message to a Second Consumer After ch_A Commits Generating UUID_B Committing ch_B, How Pulsar Native Dead-Letter Policy Routes the Message to the DLT After maxRedeliveryCount Where Re-Ingestion by an On-Call Engineer Generates UUID_B, and How PulsarAdmin.resetCursor() to the Outage Window Start Replays Already-Charged Records Committing ch_B × N

By Keybrake —

The Spring Cloud Stream Pulsar Binder has three failure modes that produce duplicate Stripe charges — and none of them come from the same mechanics as the SCS Kafka or RabbitMQ binders. Each one is rooted in something Pulsar does differently at the broker level: per-message delivery timeout tracking, native broker-side dead-lettering, and subscription cursor management via the Admin API. If you're running a Pulsar-backed billing consumer and you haven't thought through these three scenarios, a slow international payment, a routine DLT audit, or an outage-recovery cursor reset can each silently commit ch_B to Stripe.

This post covers all three end to end: the exact SCS Pulsar Binder configuration that enables each failure, the step-by-step timeline showing when ch_A commits and when UUID_B is generated, the specific misconception each one exploits, and the idempotency fix that makes your billing consumer safe regardless of which Pulsar delivery guarantee is in play. Code samples are Spring Cloud Stream 4.x with the Spring Pulsar Binder, Apache Pulsar 3.x, Spring Boot 3.x, WireMock, Testcontainers, and JUnit 5.

Background: what makes the SCS Pulsar Binder different from Kafka and RabbitMQ

Before working through the three failure modes it is worth stating what Pulsar does differently, because each failure mode is a direct consequence of one of these differences.

Per-message ack tracking with wall-clock timeout. Pulsar tracks each delivered message-ID individually in a per-subscription ack set. When the broker delivers a message, it starts a wall-clock timer parameterized by ackTimeout on the subscription. If the consumer does not acknowledge the specific message-ID before the timer fires, the broker redelivers it — to the same consumer if it is still connected, or to any available consumer in a Shared subscription. This is fundamentally different from Kafka's max.poll.interval.ms, which tracks the time between poll() calls at the consumer-group level, not the wall-clock time from delivery to ack for a specific record. It is also different from RabbitMQ's consumer timeout, which targets idle channels rather than per-message delivery duration.

Shared subscription delivers the same message to any consumer in the group. In Pulsar's Shared subscription type, the broker dispatches messages round-robin to all connected consumers on the subscription. When a message is redelivered due to ackTimeout, the broker sends it to whichever consumer has the lowest pending ack count at that moment — which may be a different consumer instance from the one that originally received it, even if the original consumer is still alive and mid-execution. Kafka's rebalance-based redelivery always waits until the original partition assignment is revoked; RabbitMQ's AMQP redelivery requeues to the same queue and any consumer picks it up. Only Pulsar Shared can deliver a redelivery to Consumer B while Consumer A is still processing the original delivery.

Native broker-side dead-lettering independent of SCS maxAttempts. Spring Cloud Stream's RabbitMQ and Kafka binders implement dead-lettering as an SCS-layer concern: the binder retries maxAttempts times in application memory and then calls basicReject (RabbitMQ) or publishes to a DLT topic (Kafka) as an application-level side effect. The SCS Pulsar Binder exposes Pulsar's native DeadLetterPolicy, which is a broker-side construct: the Pulsar broker itself tracks how many times a given message-ID has been delivered to the subscription and, after maxRedeliveryCount deliveries, automatically routes the message to the configured dead-letter topic without the application needing to call any explicit dead-letter API. This means the DLT routing happens at the Pulsar broker even if the SCS consumer function crashes mid-execution on every attempt.

Subscription cursor management via Admin API. Pulsar maintains a per-subscription read cursor as a persistent log position (a MessageId which encodes ledger, entry, and partition index). The Pulsar Admin API exposes topics().resetCursor(topic, subscription, timestamp) which repositions the subscription cursor to the earliest message whose publish time is greater than or equal to the given timestamp. Unlike Kafka's --reset-offsets --to-datetime CLI which operates on consumer group offsets at the partition level, Pulsar's resetCursor operates on the Pulsar subscription's persistent cursor state and takes effect immediately on the next message dispatch — no consumer restart required.

With that background in place, the three failure modes follow directly.

Failure mode 1: subscriptionType=Shared + ackTimeout=30000 redelivers in-flight billing message to Consumer B after Consumer A already committed ch_A

The setup

A team runs a subscription billing service on Apache Pulsar 3.3. The billing pipeline is a Spring Cloud Stream application with the SCS Pulsar Binder. The consumer function receives SubscriptionOrder messages from the persistent://default/default/billing topic, calls the Stripe Charges API, and writes a BillingRecord to PostgreSQL. Two consumer instances run as a Kubernetes Deployment so that billing throughput scales horizontally.

The developer chooses subscriptionType=Shared so that both consumer pods receive messages from the same subscription and share the workload. The developer also sets ackTimeout=30000 (30 seconds) based on a reading of the Pulsar documentation that describes ackTimeout as a safety net for stuck or lost messages — if a consumer stops responding, ackTimeout ensures the message is redelivered rather than lost.

# application.yml
spring:
  cloud:
    pulsar:
      client:
        service-url: pulsar://pulsar-broker:6650
    stream:
      pulsar:
        binder:
          service-url: pulsar://pulsar-broker:6650
        bindings:
          billing-in-0:
            consumer:
              subscription-type: Shared
              subscription-name: billing-group
              ack-timeout-millis: 30000          # 30 s — "safety net for stuck messages"
              ack-timeout-redelivery-backoff:
                min-delay-ms: 1000
                max-delay-ms: 10000
                multiplier: 2.0
      bindings:
        billing-in-0:
          destination: billing
          group: billing-group

The consumer function is straightforward:

@Configuration
public class BillingConsumer {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<Message<SubscriptionOrder>> billing() {
        return msg -> {
            SubscriptionOrder order = msg.getPayload();

            // BUG: fresh UUID per invocation — no idempotency key
            String idempotencyKey = UUID.randomUUID().toString();

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(order.getAmountCents())
                .setCurrency("usd")
                .setCustomer(order.getStripeCustomerId())
                .setDescription("Subscription " + order.getSubscriptionId())
                .build();

            // Stripe HTTP call — can take 10-20 s for 3DS / international cards
            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            // PostgreSQL write — can be slow under connection pool contention
            billingRecordRepo.save(new BillingRecord(
                order.getSubscriptionId(),
                order.getBillingPeriod(),
                charge.getId(),
                idempotencyKey
            ));
        };
    }
}

The failure timeline

A billing run processes 400 subscription orders. Message M1 carries a SubscriptionOrder for customer C1, a US-based customer paying with a card that requires Stripe Radar 3D Secure review for amounts above $500. The charge amount is $599.

Customer C1 sees two charges of $599 on their card statement. Stripe has two separate charge records, ch_A and ch_B, each under a different idempotency key. There is no indication in the Stripe dashboard that these were duplicate charges — they look like two independent transactions.

The misconception

"ackTimeout is a safety net for messages that are truly stuck — if a consumer starts processing a message and is making forward progress, the ack timeout won't fire while the consumer is alive. It's only for messages where the consumer has crashed or the JVM froze. Our consumers are healthy, so ackTimeout is harmless here."

This is wrong in a specific and important way. Pulsar's ackTimeout is a wall-clock timer that measures elapsed time from message delivery to ack receipt, with no concept of "consumer is making progress." The broker does not distinguish between a consumer that crashed at t=1ms (stuck) and a consumer that is actively calling an external API and waiting for a response (progressing). Both look identical from the broker's perspective: a delivered message with no ack. When the timer fires, the broker redelivers unconditionally. Consumer liveness is tracked separately via the consumer's heartbeat connection to the broker; ackTimeout is a message-delivery-level mechanism, not a consumer-liveness-level mechanism.

In Pulsar's Shared subscription, this redelivery goes to whichever consumer has the lowest pending-ack count — which may be a different consumer from the one that is mid-execution. The two consumers are now both processing the same message simultaneously, each with a different UUID.randomUUID() key, and Stripe will commit two charges.

The fix

There are two layers to the fix. The immediate fix is idempotency: derive the Stripe idempotency key from stable fields of the SubscriptionOrder payload, not from UUID.randomUUID(). With a stable key, Consumer B's Stripe call will be deduplicated against Consumer A's call, and Stripe will return ch_A from its idempotency cache. The second layer is a guard-before-Stripe check: query BillingRecord for the stable key before calling Stripe so that a redelivered consumer skips the Stripe call entirely if the charge was already committed.

@Configuration
public class BillingConsumer {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Bean
    public Consumer<Message<SubscriptionOrder>> billing() {
        return msg -> {
            SubscriptionOrder order = msg.getPayload();

            // Stable key from payload fields — same value on every redelivery
            String idempotencyKey = sha256(
                order.getSubscriptionId() + "|" +
                order.getBillingPeriod() + "|" +
                order.getAmountCents()
            );

            // Guard: if BillingRecord already exists for this key, skip Stripe entirely
            if (billingRecordRepo.existsByIdempotencyKey(idempotencyKey)) {
                return; // ack without calling Stripe — ch_A already committed
            }

            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(order.getAmountCents())
                .setCurrency("usd")
                .setCustomer(order.getStripeCustomerId())
                .setDescription("Subscription " + order.getSubscriptionId())
                .build();

            Charge charge = stripe.charges().create(params,
                RequestOptions.builder()
                    .setIdempotencyKey(idempotencyKey)
                    .build());

            billingRecordRepo.save(new BillingRecord(
                order.getSubscriptionId(),
                order.getBillingPeriod(),
                charge.getId(),
                idempotencyKey
            ));
        };
    }

    private static String sha256(String input) {
        try {
            MessageDigest digest = MessageDigest.getInstance("SHA-256");
            byte[] hash = digest.digest(input.getBytes(StandardCharsets.UTF_8));
            return HexFormat.of().formatHex(hash);
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException(e);
        }
    }
}

Additionally, set ackTimeout to a value that is larger than your worst-case Stripe p99 plus worst-case DB write p99 plus a safety margin. If your Stripe 3DS p99 is 15 seconds and your PostgreSQL p99 under pool contention is 20 seconds, set ackTimeout to at least 60,000ms (60 seconds). With a stable idempotency key, even if ackTimeout fires and redelivers to a second consumer, the second Stripe call will return ch_A from the cache rather than committing ch_B.

# application.yml — fixed ackTimeout value
spring:
  cloud:
    stream:
      pulsar:
        bindings:
          billing-in-0:
            consumer:
              subscription-type: Shared
              subscription-name: billing-group
              # Set higher than Stripe p99 + DB write p99 + safety margin
              ack-timeout-millis: 120000     # 120 s
              ack-timeout-redelivery-backoff:
                min-delay-ms: 5000
                max-delay-ms: 30000
                multiplier: 2.0

Test: Stripe called exactly once despite ackTimeout redelivery to a second consumer

@SpringBootTest
@Testcontainers
class BillingAckTimeoutIdempotencyTest {

    @Container
    static PulsarContainer pulsar = new PulsarContainer(
        DockerImageName.parse("apachepulsar/pulsar:3.3.1"));

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
        DockerImageName.parse("postgres:16-alpine"));

    @RegisterExtension
    static WireMockExtension wireMock = WireMockExtension.newInstance()
        .options(wireMockConfig().dynamicPort())
        .build();

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.cloud.pulsar.client.service-url", pulsar::getPulsarBrokerUrl);
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("stripe.base-url", () -> wireMock.baseUrl());
    }

    @Test
    void stripeCalledOnceWhenAckTimeoutRedeliversToSecondConsumer() throws Exception {
        String subscriptionId = "sub_test_001";
        String billingPeriod  = "2026-10";
        long   amountCents    = 59900L;
        String expectedKey    = sha256(subscriptionId + "|" + billingPeriod + "|" + amountCents);

        // First Stripe call — slow (simulate 3DS delay) but succeeds with ch_A
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .withHeader("Idempotency-Key", equalTo(expectedKey))
            .willReturn(aResponse()
                .withFixedDelay(13_000)          // 13 s — triggers ackTimeout at 30 s threshold
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody(chargeJson("ch_A", 59900L, "succeeded"))));

        // Any subsequent call with same key returns ch_A from idempotency cache
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .withHeader("Idempotency-Key", equalTo(expectedKey))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody(chargeJson("ch_A", 59900L, "succeeded"))));

        // Produce one billing message — ackTimeout will cause redelivery to second consumer
        try (PulsarClient client = PulsarClient.builder()
                 .serviceUrl(pulsar.getPulsarBrokerUrl()).build();
             Producer<byte[]> producer = client.newProducer()
                 .topic("persistent://public/default/billing").create()) {

            producer.send(objectMapper.writeValueAsBytes(
                new SubscriptionOrder(subscriptionId, billingPeriod, amountCents, "cus_stripe_001")));
        }

        // Wait for both consumers to have processed (ackTimeout fires at 30 s)
        Thread.sleep(45_000);

        // Stripe should have been called exactly once — second call returns cached ch_A
        wireMock.verify(exactly(1), postRequestedFor(urlEqualTo("/v1/charges")));

        // BillingRecord should exist exactly once
        long count = billingRecordRepo.countByIdempotencyKey(expectedKey);
        assertThat(count).isEqualTo(1);
    }
}

Failure mode 2: Pulsar native DeadLetterPolicy.maxRedeliveryCount=5 routes billing message to DLT — on-call engineer reads DataIntegrityViolationException in DLT message properties, concludes Stripe was never called, re-ingests message, generates UUID_B, commits ch_B

The setup

The team extends the billing service to capture permanently failed messages using Pulsar's native DeadLetterPolicy. The intent is to give on-call engineers a way to inspect and manually re-process billing messages that exhausted all retry attempts.

# application.yml
spring:
  cloud:
    stream:
      pulsar:
        binder:
          service-url: pulsar://pulsar-broker:6650
        bindings:
          billing-in-0:
            consumer:
              subscription-name: billing-group
              subscription-type: Exclusive           # one consumer, no ackTimeout confusion
              dead-letter-policy:
                max-redelivery-count: 5              # Pulsar broker routes to DLT after 5 deliveries
                dead-letter-topic: >-
                  persistent://default/default/billing-dlq
                retry-letter-topic: >-
                  persistent://default/default/billing-retry
              retry-enable: true
      bindings:
        billing-in-0:
          destination: billing
          group: billing-group

The Pulsar-native DeadLetterPolicy differs from the SCS-layer maxAttempts retry in a critical way: the Pulsar broker tracks per-message delivery count independently of what the application does. SCS Pulsar Binder's internal retry (maxAttempts) retries in application memory within a single broker delivery. Once all SCS retries are exhausted for one delivery, the binder negativeAcks the message. The broker increments the delivery count for that message-ID. After 5 broker-level deliveries, the broker routes to the dead-letter topic — regardless of how many SCS application-level retries happened within each delivery. In this configuration the billing message will reach the DLT after 5 × maxAttempts total application invocations.

The consumer records the last exception in a message property that the team's monitoring system attaches to the DLT message via a custom error handler:

@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> errorHandler() {
    return (container, dest, group) -> {
        container.setupMessageListener((MessageListener<byte[]>) msg -> {
            try {
                processBillingMessage(msg);
                msg.acknowledge();
            } catch (Exception ex) {
                // Record the exception class on the message for DLT inspection
                msg.getProperties().put("x-last-exception", ex.getClass().getSimpleName());
                msg.negativeAcknowledge();
            }
        });
    };
}

The failure timeline

Billing message M2 for customer C2 enters the pipeline. C2's subscription has a known-bad idempotency scenario: the billing period timestamp edge case causes a unique constraint violation in BillingRecord because a previous billing run left a partial record with the same (subscriptionId, billingPeriod) composite key.

The misconception

"The DLT message properties show x-last-exception: DataIntegrityViolationException. A DataIntegrityViolationException is a database exception. If the last thing that failed was a database write, that means the consumer function was executing the database write when it died — which means Stripe was called before the database, so Stripe must have already committed, but wait, no — the code calls Stripe first and then the database, so if the database threw, Stripe must have... hmm. Actually the code calls the database first to validate, then Stripe, then saves. So DataIntegrityViolationException means the pre-validation DB check failed, meaning Stripe was never reached."

This is a rationalisation error: the engineer constructs a story about code ordering that matches the exception but doesn't match the actual code. The real code calls Stripe first (stripe.charges().create()), then calls billingRecordRepo.save(). The DataIntegrityViolationException is thrown by the DB save, which comes after Stripe. But because the exception is the only signal visible in the DLT message, the engineer reverse-engineers a code ordering that explains the exception as a pre-Stripe failure — and that reverse-engineering is wrong.

There is a deeper problem: even if the engineer correctly knew the code ordering (Stripe first, then DB), the engineer might still reason: "Stripe committed on attempt 1. The DLT exception is from attempt 5. Between attempts 1 and 5, all the re-attempts also called Stripe. So the DLT message is already associated with a committed charge — do not re-ingest." A correct understanding of the DLT mechanics would lead to the right conclusion: a DLT message almost always means Stripe was called at least once — re-ingest only if your consumer is idempotent.

The fix

Fix the root cause first: why does billingRecordRepo.save() throw DataIntegrityViolationException on every attempt? If it is a stale partial record, that record must be cleaned up before billing can succeed. But fixing the root cause alone is not sufficient — you must also make the consumer idempotent so that past and future DLT re-ingestions are safe regardless of the exception recorded in the DLT message.

@Bean
public Consumer<Message<SubscriptionOrder>> billing() {
    return msg -> {
        SubscriptionOrder order = msg.getPayload();

        String idempotencyKey = sha256(
            order.getSubscriptionId() + "|" +
            order.getBillingPeriod() + "|" +
            order.getAmountCents()
        );

        // Guard: skip Stripe if we already have a BillingRecord for this key
        Optional<BillingRecord> existing =
            billingRecordRepo.findByIdempotencyKey(idempotencyKey);
        if (existing.isPresent()) {
            // Already charged — acknowledge without calling Stripe
            return;
        }

        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount(order.getAmountCents())
            .setCurrency("usd")
            .setCustomer(order.getStripeCustomerId())
            .build();

        Charge charge = stripe.charges().create(params,
            RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build());

        // Use upsert semantics to handle the DataIntegrityViolationException root cause
        billingRecordRepo.upsert(new BillingRecord(
            order.getSubscriptionId(),
            order.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    };
}

With this fix, a re-ingested DLT message produces the following outcome: the consumer computes the same stable key as all previous delivery attempts. It queries BillingRecord for that key. If any previous delivery attempt succeeded in writing the record (after the upsert fix is deployed), the re-ingestion returns immediately. If no record exists (because all previous attempts failed at the DB write before the upsert fix), the consumer calls Stripe. Stripe returns ch_A (or whichever charge was last committed under the idempotency cache for this key) and the upsert write succeeds. One charge total, regardless of how many times the DLT message was re-ingested.

Also add a team runbook note: never re-ingest a DLT billing message from a consumer that calls Stripe unless the consumer is verified idempotent. The exception type in the DLT message properties tells you what failed on the last attempt, not whether Stripe was called.

DLT message property reference for Pulsar native dead-letter policy

Property Value Meaning
__MessageDeathCount__ integer How many times this message-ID was delivered to the subscription before DLT routing. Does NOT indicate Stripe was called 0 times.
__OriginMessageIdData__ base64-encoded MessageId The original message-ID in the source topic. Use this to correlate the DLT message with source topic position.
__OriginTopicName__ topic URL string The topic the message came from. Use this to verify re-ingestion target.
Custom x-last-exception exception class name Set by application error handler. Reflects the last exception caught — NOT the exception from delivery 1. May be a post-Stripe exception even if Stripe succeeded on delivery 1.

Test: Stripe called exactly once despite DLT re-ingestion

@SpringBootTest
@Testcontainers
class BillingDltReIngestionIdempotencyTest {

    @Container
    static PulsarContainer pulsar = new PulsarContainer(
        DockerImageName.parse("apachepulsar/pulsar:3.3.1"));

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
        DockerImageName.parse("postgres:16-alpine"));

    @RegisterExtension
    static WireMockExtension wireMock = WireMockExtension.newInstance()
        .options(wireMockConfig().dynamicPort())
        .build();

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.cloud.pulsar.client.service-url", pulsar::getPulsarBrokerUrl);
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("stripe.base-url", () -> wireMock.baseUrl());
        // Set maxRedeliveryCount=2 so test completes faster than maxRedeliveryCount=5
        registry.add("spring.cloud.stream.pulsar.bindings.billing-in-0.consumer.dead-letter-policy.max-redelivery-count", () -> "2");
    }

    @Test
    void stripeCalledOnceWhenDltMessageIsReIngested() throws Exception {
        String subscriptionId = "sub_dlt_001";
        String billingPeriod  = "2026-10";
        long   amountCents    = 4900L;
        String expectedKey    = sha256(subscriptionId + "|" + billingPeriod + "|" + amountCents);

        // Stripe stub: returns ch_A for the stable key; second call returns cached ch_A
        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .withHeader("Idempotency-Key", equalTo(expectedKey))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody(chargeJson("ch_A", amountCents, "succeeded"))));

        // Produce to billing topic — will exhaust maxRedeliveryCount and land in DLT
        // (In this test the consumer succeeds on first attempt with the stable key fix,
        //  so the DLT is never reached — but we simulate re-ingestion by producing again)
        try (PulsarClient client = PulsarClient.builder()
                 .serviceUrl(pulsar.getPulsarBrokerUrl()).build();
             Producer<byte[]> producer = client.newProducer()
                 .topic("persistent://public/default/billing").create()) {

            byte[] payload = objectMapper.writeValueAsBytes(
                new SubscriptionOrder(subscriptionId, billingPeriod, amountCents, "cus_stripe_001"));

            // Original delivery
            producer.send(payload);
            Thread.sleep(3_000);

            // Simulate on-call engineer re-ingesting from DLT back to billing topic
            producer.send(payload);
            Thread.sleep(3_000);

            // Simulate a second re-ingestion for good measure
            producer.send(payload);
            Thread.sleep(3_000);
        }

        // Stripe must have been called exactly once — re-ingestions hit the BillingRecord guard
        wireMock.verify(exactly(1), postRequestedFor(urlEqualTo("/v1/charges")));

        // BillingRecord must exist exactly once for this key
        assertThat(billingRecordRepo.countByIdempotencyKey(expectedKey)).isEqualTo(1);
    }
}

Failure mode 3: PulsarAdmin.topics().resetCursor() to the outage window start replays already-charged billing records — UUID_B × N — ch_B × N

The setup

The billing service runs in production with a stable idempotent consumer (the stable-key fix has been applied). On October 23, 2026, a PostgreSQL primary failover at 08:02 UTC causes a 4-minute outage window from 08:02 UTC to 08:06 UTC. During this window the billing consumer receives 2,400 messages but all 2,400 billingRecordRepo.save() calls throw DataIntegrityViolationException or JdbcSQLTimeoutException (PostgreSQL connection refused). The Stripe charges for these 2,400 messages do commit — the Stripe HTTP call happens before the DB write, and Stripe was not affected by the PostgreSQL outage. But the consumers negativeAck all 2,400 messages because the DB write failed. The Pulsar broker requeues them.

When the PostgreSQL primary is restored at 08:06 UTC, the consumer resumes and begins reprocessing the 2,400 negativeAcked messages. Because the consumer is idempotent (stable key + BillingRecord guard), the reprocessing is safe: each guard query finds no existing BillingRecord (the DB writes failed), calls Stripe with the same key, Stripe returns the same charge from its idempotency cache, and the DB write succeeds. The 2,400 billing records are committed correctly.

However, the on-call engineer monitoring the incident does not know about this automatic reprocessing. The engineer's incident runbook says: "after a PostgreSQL outage, reset the billing consumer subscription cursor to the outage start time to ensure all messages from the outage window are reprocessed." The engineer follows the runbook step by step.

The failure

At 08:15 UTC the engineer runs:

# Pulsar Admin CLI — reset subscription cursor to outage start
pulsar-admin topics reset-cursor \
    --subscription billing-group \
    --time "2026-10-23T08:00:00.000Z" \
    persistent://default/default/billing

The engineer chose 08:00 UTC — 2 minutes before the outage — as a safety buffer, reasoning that "a 2-minute buffer ensures we don't miss any messages that were in-flight when the outage hit."

What the engineer does not know:

The resetCursor command repositions the billing-group subscription cursor to the earliest message whose publish time is ≥ 08:00 UTC, which is the start of the 800-message pre-outage window. From this point forward, the billing consumer reads from this position as if those messages were unprocessed.

At 08:15 UTC the consumer starts receiving the replayed 800 messages from the 08:00–08:02 UTC window. With the idempotent consumer (stable key + BillingRecord guard), the guard query returns an existing BillingRecord for each message and the consumer acknowledges without calling Stripe. The 800 customers are not double-charged. The fix works.

But in the scenario where the idempotency fix has not been applied (the original UUID.randomUUID() consumer), the cursor reset causes a full double-charge of all 800 pre-outage customers plus a re-charge of the 2,400 outage-window customers (which the auto-reprocessing had already handled). Total double-charges: up to 3,200 customers.

The misconception

"resetCursor resets the subscription cursor to a position in the topic, but Pulsar tracks acknowledged message IDs in a persistent ack set per subscription. Messages that were already acknowledged before the cursor reset are in the ack set. When the consumer reads from the reset position and encounters a previously-acknowledged message ID, it skips it automatically — Pulsar's individual acknowledgment model means the ack set always takes precedence over the cursor position."

This is an understandable misconception about Pulsar's individual acknowledgment model. Pulsar does track acknowledged message IDs in a per-subscription ack set, and individual acknowledgment does allow a subscription to acknowledge message IDs out of order. But resetCursor does not work alongside the individual ack set the way the engineer imagines.

When you call resetCursor, the Pulsar broker does two things: (1) it repositions the subscription's read cursor to the target position, and (2) it clears the individual ack set for the subscription. The ack set is tied to the cursor range: once the cursor is moved backward, the ack set for the range ahead of the new cursor position is cleared because the broker needs to track which messages in that range are acknowledged going forward. Previously-acknowledged message IDs in the rolled-back range are no longer marked acknowledged.

This means: after resetCursor to 08:00 UTC, the 800 messages from 08:00–08:02 UTC that were previously acknowledged are now unacknowledged as far as the broker is concerned. They will be delivered to the consumer again. If the consumer is not idempotent, they will call Stripe again.

Pulsar subscription cursor vs Kafka offset reset: key differences

Dimension Pulsar resetCursor Kafka --reset-offsets --to-datetime
Ack set behavior on reset Clears individual ack set for the rolled-back range — previously-acked messages are re-delivered Committed offsets are overwritten — all records at or after the new offset are re-consumed
Requires consumer restart No — takes effect immediately on next broker dispatch Yes — consumer group must be inactive (no running consumers)
API granularity Per subscription on a topic Per consumer group, per partition
Admin interface PulsarAdmin.topics().resetCursor(topic, subscription, timestamp) kafka-consumer-groups.sh --reset-offsets --to-datetime --execute
Double-charge risk without idempotency High — acked messages in rolled-back range are re-delivered High — same behavior from Kafka's perspective

The fix

The billing consumer must be idempotent before any cursor-based operational recovery is safe. With the stable key + BillingRecord guard in place, resetCursor is safe because the guard query short-circuits Stripe for any message that was already processed. The cursor reset becomes an idempotent replay operation.

Additionally, update the runbook to check for automatic reprocessing before manually resetting the cursor:

# RUNBOOK: Billing consumer recovery after PostgreSQL outage
# Step 1: Verify automatic reprocessing completed
pulsar-admin topics stats persistent://default/default/billing \
    | jq '.subscriptions["billing-group"].msgBacklog'
# If backlog is 0: automatic reprocessing succeeded — do NOT reset cursor
# If backlog > 0: messages are still queued — verify consumer is running and processing

# Step 2 (only if manual reset is required and consumer is verified idempotent):
# Record the exact outage start time from your incident timeline
OUTAGE_START="2026-10-23T08:02:00.000Z"   # Use exact outage time, not a buffer

pulsar-admin topics reset-cursor \
    --subscription billing-group \
    --time "${OUTAGE_START}" \
    persistent://default/default/billing

# WARNING: resetCursor clears the ack set for the rolled-back range.
# Any message that was acknowledged in the window [OUTAGE_START, now) will be redelivered.
# This is safe ONLY if the consumer has a stable idempotency key + BillingRecord guard.

The Java equivalent for programmatic recovery:

// Programmatic cursor reset via Pulsar Admin Java client
try (PulsarAdmin admin = PulsarAdmin.builder()
        .serviceHttpUrl("http://pulsar-admin:8080")
        .build()) {

    String topic        = "persistent://default/default/billing";
    String subscription = "billing-group";

    // Use exact outage start time — no safety buffer
    // A buffer redelivers already-processed messages even with idempotency fix
    long outageStartEpochMs = Instant.parse("2026-10-23T08:02:00.000Z")
        .toEpochMilli();

    admin.topics().resetCursor(topic, subscription, outageStartEpochMs);
    log.info("Subscription cursor reset to {}. Consumer must be idempotent.", outageStartEpochMs);
}

Test: Stripe called exactly once despite cursor reset that redelivers already-charged records

@SpringBootTest
@Testcontainers
class BillingCursorResetIdempotencyTest {

    @Container
    static PulsarContainer pulsar = new PulsarContainer(
        DockerImageName.parse("apachepulsar/pulsar:3.3.1"));

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
        DockerImageName.parse("postgres:16-alpine"));

    @RegisterExtension
    static WireMockExtension wireMock = WireMockExtension.newInstance()
        .options(wireMockConfig().dynamicPort())
        .build();

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    @DynamicPropertySource
    static void properties(DynamicPropertyRegistry registry) {
        registry.add("spring.cloud.pulsar.client.service-url", pulsar::getPulsarBrokerUrl);
        registry.add("spring.datasource.url", postgres::getJdbcUrl);
        registry.add("stripe.base-url", () -> wireMock.baseUrl());
    }

    @Test
    void stripeCalledOnceWhenCursorResetCausesRedeliveryOfAlreadyChargedMessages()
            throws Exception {

        String subscriptionId = "sub_cursor_001";
        String billingPeriod  = "2026-10";
        long   amountCents    = 2999L;
        String expectedKey    = sha256(subscriptionId + "|" + billingPeriod + "|" + amountCents);

        wireMock.stubFor(post(urlEqualTo("/v1/charges"))
            .withHeader("Idempotency-Key", equalTo(expectedKey))
            .willReturn(aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody(chargeJson("ch_A", amountCents, "succeeded"))));

        String topic        = "persistent://public/default/billing";
        String subscription = "billing-group";

        // Step 1: produce and process one billing message (charge committed, acked)
        long beforeProduce = Instant.now().toEpochMilli();

        try (PulsarClient client = PulsarClient.builder()
                 .serviceUrl(pulsar.getPulsarBrokerUrl()).build();
             Producer<byte[]> producer = client.newProducer()
                 .topic(topic).create()) {
            producer.send(objectMapper.writeValueAsBytes(
                new SubscriptionOrder(subscriptionId, billingPeriod, amountCents, "cus_stripe_001")));
        }

        // Wait for consumer to process and ack
        Thread.sleep(3_000);
        assertThat(billingRecordRepo.countByIdempotencyKey(expectedKey)).isEqualTo(1);

        // Step 2: reset cursor to before the message was produced (simulates engineer recovery)
        try (PulsarAdmin admin = PulsarAdmin.builder()
                .serviceHttpUrl(pulsar.getHttpServiceUrl()).build()) {
            admin.topics().resetCursor(topic, subscription, beforeProduce);
        }

        // Step 3: wait for the redelivered message to be processed
        Thread.sleep(3_000);

        // Stripe must have been called exactly once — cursor-reset redelivery hits BillingRecord guard
        wireMock.verify(exactly(1), postRequestedFor(urlEqualTo("/v1/charges")));

        // BillingRecord still exactly one row for this key
        assertThat(billingRecordRepo.countByIdempotencyKey(expectedKey)).isEqualTo(1);
    }
}

SCS Pulsar Binder + Stripe billing: 8-item diagnostic checklist

When investigating a double-charge incident in a Spring Cloud Stream Pulsar Binder application, work through this checklist before touching Stripe or the database.

  1. Check ackTimeout vs Stripe p99. Run pulsar-admin topics stats --get-precise-backlog <topic> and compare the subscription's msgRateRedeliver with your Stripe p99. If ackTimeout < Stripe p99 + DB write p99, ackTimeout redelivery can fire before the original consumer finishes. This is the first thing to check in a Shared subscription setup.
  2. Count unique Stripe idempotency keys per billing message. Query Stripe: stripe charges list --limit 10 --created[gte]=<incident start>. If you see two charges with different idempotency keys for the same customer and same amount within minutes of each other, the consumer is using non-stable idempotency keys (UUID.randomUUID()) and was called twice for the same logical message.
  3. Inspect DLT message properties before re-ingesting. Read x-last-exception, __MessageDeathCount__, and the original payload. Do not assume the exception type tells you whether Stripe was called. Always check the BillingRecord table for the stable key derived from the payload before re-ingesting any DLT message.
  4. Check whether automatic negativeAck reprocessing completed before resetting cursor. Run pulsar-admin topics stats <topic> | jq '.subscriptions[<sub>].msgBacklog'. If backlog is 0, the consumer already reprocessed the outage-window messages — do not reset cursor.
  5. Verify the stable idempotency key produces the same value for the same payload. Log the computed key at DEBUG level. Check that Consumer A and Consumer B (in a Shared subscription with concurrent redelivery) would compute the same key for the same SubscriptionOrder. If the key includes a timestamp or session ID, it is not stable.
  6. Check BillingRecord existence before any operational recovery. For any message you are about to re-produce, re-ingest from DLT, or trigger via cursor reset — query SELECT count(*) FROM billing_record WHERE idempotency_key = <stable_key>. If count > 0, Stripe was already called and committed. Do not call Stripe again.
  7. Verify cursor reset cleared the ack set. After a resetCursor, run pulsar-admin topics stats --get-precise-backlog <topic> again. The backlog will reflect the number of messages between the new cursor position and the current end of the topic — including messages that were previously acknowledged. This count tells you how many messages will be redelivered. If that number is larger than your outage-window backlog, you have rolled back past already-processed messages.
  8. Use Stripe idempotency cache as a safety net, not a primary guard. The Stripe idempotency cache deduplicates calls with the same key within 24 hours (for most endpoints) or up to 6 months for some. If your redelivery or re-ingestion happens more than 24 hours after the original charge, the Stripe cache may have expired and a re-call with the same key will commit a new charge. Do not rely on the Stripe cache alone — use the BillingRecord guard as the primary deduplication mechanism.

Pulsar subscription type comparison for billing consumers

Subscription type Message distribution ackTimeout redelivery target Double-charge risk without idempotency
Exclusive All messages to single consumer Same consumer (if connected) or fails if consumer disconnects Low from concurrent delivery; still present on reconnect
Shared Round-robin across all consumers Any available consumer in subscription High — ackTimeout can deliver to Consumer B while Consumer A is still running
Key_Shared Messages with same key always to same consumer The consumer owning that key range (same consumer if connected) Medium — redelivery goes to same consumer unless ownership changes on reconnect
Failover All messages to active consumer; standby takes over on failure Standby consumer (different instance) Medium — failover redelivers in-flight messages to standby; overlapping processing window

For billing consumers, Key_Shared with key derived from subscriptionId provides better isolation than Shared: messages for the same subscription always route to the same consumer, so ackTimeout redelivery goes to the same consumer pod. This does not eliminate the double-charge risk if the pod restarts mid-execution, but it prevents the simultaneous-processing-on-different-pods scenario that Shared enables. Regardless of subscription type, a stable idempotency key plus BillingRecord guard is the only reliable protection.

The one-line summary

The SCS Pulsar Binder has three mechanics that produce duplicate Stripe charges in ways the Kafka and RabbitMQ binders do not: ackTimeout in a Shared subscription can deliver the same message to Consumer B while Consumer A is still calling Stripe; the native DLT message properties record the last exception (not the first), misleading engineers into re-ingesting already-charged messages; and PulsarAdmin.resetCursor() clears the ack set for the rolled-back range, re-delivering previously-acknowledged messages. A stable SHA-256(subscriptionId + billingPeriod + amountCents) idempotency key combined with a guard-before-Stripe query makes all three scenarios safe — ackTimeout redelivery returns the cached Stripe charge, DLT re-ingestion is a no-op if the BillingRecord exists, and cursor-reset replay hits the guard and acknowledges without calling Stripe.

Stop a runaway agent before it redelivers billing messages

Keybrake puts a spend cap and per-vendor allowlist between your agent and Stripe. One kill-switch, full audit log, no code changes required.