Keybrake

Spring Cloud Stream RocketMQ Stripe Integration: How SCS RocketMQ Binder RECONSUME_LATER Routes the In-Flight Billing Message to %RETRY%billingGroup After ch_A Commits Generating UUID_B Committing ch_B, How %DLQ%billingGroup Re-Ingestion by On-Call Engineers Generates UUID_B After ch_A Is Already Committed, and How mqadmin resetOffsetByTime to the Outage Window Start Replays Already-Charged Records Committing ch_B × N

By Keybrake —

The Spring Cloud Stream RocketMQ Binder has three failure modes that produce duplicate Stripe charges, and they arise from RocketMQ broker semantics that differ structurally from Apache Kafka, RabbitMQ, and Apache Pulsar. RocketMQ’s approach to failed-message re-delivery routes messages through named internal topics (%RETRY% and %DLQ%), its administrative offset reset tool (mqadmin resetOffsetByTime) operates on broker-stored queue offsets rather than a per-message consumed-set, and its default concurrent message listener model returns a broker-level status code rather than acknowledging or rejecting at the message-ID level. Each of these behaviors can silently commit ch_B against a Stripe customer who was already charged ch_A seconds or minutes earlier.

This post covers all three end to end: the exact SCS RocketMQ 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 about RocketMQ’s delivery model, and the idempotency fix that makes your billing consumer safe regardless of how the RocketMQ broker delivers or retries the message. Code samples are Spring Cloud Stream 4.x, Spring Cloud Stream RocketMQ Binder 4.x, Apache RocketMQ 5.x, Spring Boot 3.x, WireMock, Testcontainers, and JUnit 5.

Background: how the SCS RocketMQ Binder handles failed messages differently from Kafka

To understand why these three failure modes exist, you need to understand the two structural differences between RocketMQ’s delivery model and Kafka’s.

RocketMQ routes failed messages to a separate broker-managed retry topic, not back to the original partition. When a Kafka consumer throws an exception and the SCS Kafka Binder retries the message, the retry happens entirely within the consumer application: the binder catches the exception, applies a back-off, and re-invokes the message handler with the same ConsumerRecord that it already holds in memory. The Kafka broker is not involved in the retry loop at all. In contrast, when an SCS RocketMQ Binder consumer’s MessageListenerConcurrently returns ConsumeConcurrentlyStatus.RECONSUME_LATER, the RocketMQ broker receives that status code, increments the message’s reconsumeTimes property, and routes a copy of the message to an internal topic named %RETRY%<consumerGroup>. The consumer group subscribes to this retry topic automatically via the binder. After a delay determined by the reconsumeTimes level (10 seconds for level 3, which corresponds to the first retry), the broker delivers the message from the retry topic to the consumer.

This means that every retry is a new delivery from the broker, not a re-execution within the same listener invocation. The consumer has no persistent state linking the retry delivery to the original delivery. Any value computed locally during the original delivery — including an idempotency key generated with UUID.randomUUID() — is gone. The retry delivery starts completely fresh.

RocketMQ’s offset store is queue-positional, not message-ID-based. Both RocketMQ and Kafka track consumer progress via a numeric offset per queue (RocketMQ) or partition (Kafka). Neither maintains a log of which specific message IDs were consumed by a given consumer group. When you reset the consumer group offset — via mqadmin resetOffsetByTime in RocketMQ or kafka-consumer-groups.sh --reset-offsets in Kafka — you are repositioning the read pointer, not querying a consumed-message registry. Any message whose queue position falls at or after the new offset will be redelivered, regardless of whether the consumer group previously acknowledged it.

With these two structural differences as context, the three failure modes follow directly.

Failure mode 1: ConsumeConcurrentlyStatus.RECONSUME_LATER after ch_A commits — RocketMQ routes message to %RETRY%billing-group and redelivers after 10 seconds generating UUID_B committing ch_B

The setup

A team runs a SaaS subscription billing service for a regional market. The message broker is Apache RocketMQ 5.x. The Spring Boot 3.x billing service uses the Spring Cloud Stream RocketMQ Binder to consume billing events from the billing topic. The team configures a modest retry limit:

# application.yml
spring:
  cloud:
    stream:
      bindings:
        billing-in-0:
          destination: billing
          group: billing-group
          consumer:
            max-attempts: 1          # SCS application-level retries disabled
      rocketmq:
        binder:
          name-server: rocketmq-nameserver:9876
        bindings:
          billing-in-0:
            consumer:
              maxReconsumeTimes: 3   # RocketMQ broker-level retries
              tags: billing

The developer sets spring.cloud.stream.bindings.billing-in-0.consumer.max-attempts=1 to disable SCS application-level retries, intending to rely on RocketMQ’s built-in retry mechanism. The RocketMQ binder’s maxReconsumeTimes=3 controls how many times the broker will re-enqueue the message in %RETRY%billing-group before routing it to %DLQ%billing-group.

The billing consumer is straightforward:

@Component
public class BillingConsumer implements Consumer<BillingEvent> {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Override
    public void accept(BillingEvent event) {

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

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

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

        // Throws DataIntegrityViolationException if a duplicate row exists
        billingRecordRepo.save(new BillingRecord(
            event.getSubscriptionId(),
            event.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    }
}

The SCS RocketMQ binder wraps this consumer in a MessageListenerConcurrently. When the consumer method throws an exception, the binder returns ConsumeConcurrentlyStatus.RECONSUME_LATER to the RocketMQ broker. The developer knows this and considers it the correct behavior — the message will be retried.

The failure timeline

During a monthly billing run for 1,400 customers, message M1 carries a BillingEvent for subscription sub_RM001, a customer paying EUR 149/month.

Customer charged three times: ch_A, ch_B, ch_C. Total: EUR 447 against EUR 149 owed.

The Stripe dashboard shows three charges for the same customer with the same amount within a 41-second window, each under a different charge ID and a different idempotency key. The idempotency keys are unrelated UUID v4 strings with no connection to the subscription ID.

The misconception

“RECONSUME_LATER is RocketMQ’s equivalent of Kafka consumer retry — the message stays in the original topic partition and is retried in-place by the consumer. The broker just signals that the consumer will try again later, similar to how Kafka moves the fetch pointer back. The retry is logically within the same delivery attempt, just delayed.”

This is wrong in the detail that matters most. Kafka binder retries are entirely application-side. When the SCS Kafka Binder retries, it re-invokes the message handler from the application process, with the same ConsumerRecord object already in memory, without any broker involvement. The Kafka broker does not know that a retry is happening. The consumer application is free to pass any state from the original invocation into the retry — including an idempotency key computed during the first attempt.

RocketMQ’s RECONSUME_LATER is a fundamentally different mechanism. The broker receives the status code, takes ownership of the message, re-enqueues it in %RETRY%<consumerGroup>, and delivers it as a new message after the delay interval. The consumer application’s in-memory state from the original delivery — local variables, idempotency keys, partial results — does not survive this round-trip. Every retry delivery is a fresh accept() invocation with no connection to the original invocation except the message body.

The RocketMQ default delay levels for RECONSUME_LATER are worth memorizing because they affect how long the window is between the original Stripe charge and the duplicate charge:

reconsumeTimes Delay level Wait before redelivery Time since ch_A commits (at t=500ms)
0 → 1 (first retry) 3 10 seconds ~10.5 seconds
1 → 2 (second retry) 4 30 seconds ~41 seconds
2 → 3 (third retry) 5 1 minute ~101 seconds
3 → 4 (fourth retry) 6 2 minutes ~221 seconds
15 → 16 (sixteenth retry) 18 2 hours ~7,201 seconds

Note that the default maxReconsumeTimes in the SCS RocketMQ Binder is -1, which maps to RocketMQ’s broker default of 16 retry attempts. With UUID.randomUUID() and no idempotency guard, every retry attempt is a new Stripe charge.

Diagnosing this failure in production

The signature in your application logs is: the initial DataIntegrityViolationException, followed by silence for approximately 10 seconds, followed by a repeat of the same exception. If you are logging the Stripe charge ID, you will see different charge IDs on each attempt despite identical message bodies. If you are not logging the charge ID, the duplicate charges will only be visible in Stripe’s dashboard.

In the RocketMQ console or via mqadmin printMsg -t %RETRY%billing-group, you will see the message with incrementing reconsumeTimes values. The message body will be identical across all entries in the retry topic.

The fastest diagnostic SQL query once you suspect a billing run has duplicate charges:

-- Find subscriptions with more than one billing record per period
SELECT subscription_id, billing_period, COUNT(*) as charge_count,
       ARRAY_AGG(stripe_charge_id ORDER BY created_at) as charges
FROM billing_records
WHERE billing_period = '2026-10'
GROUP BY subscription_id, billing_period
HAVING COUNT(*) > 1
ORDER BY charge_count DESC;

The fix

Generate the idempotency key from stable payload fields before the Stripe call, and add a guard that checks for an existing BillingRecord before calling Stripe. Both defenses are necessary: the stable key ensures Stripe’s own idempotency cache deduplicates concurrent or near-concurrent calls with the same key, and the guard catches cases where the Stripe cache has expired (it has a 30-minute window) or where the key was generated differently on different attempts.

@Component
public class BillingConsumer implements Consumer<BillingEvent> {

    private final StripeClient stripe;
    private final BillingRecordRepository billingRecordRepo;

    @Override
    public void accept(BillingEvent event) {

        // Stable key: deterministic from payload — identical on every retry
        String idempotencyKey = sha256(
            event.getSubscriptionId() + "|" +
            event.getBillingPeriod()  + "|" +
            event.getAmountCents()
        );

        // Guard: skip Stripe if this billing period was already committed
        if (billingRecordRepo.existsBySubscriptionIdAndBillingPeriod(
                event.getSubscriptionId(), event.getBillingPeriod())) {
            return; // idempotent: already processed, ack without re-charging
        }

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

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

        billingRecordRepo.save(new BillingRecord(
            event.getSubscriptionId(),
            event.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);
        }
    }
}

With the stable key, every RECONSUME_LATER retry recomputes the same idempotency key from the same event payload. When the message is redelivered from %RETRY%billing-group, the Stripe POST /v1/charges call carries Idempotency-Key: <SHA-256_A>. Stripe returns the cached ch_A response without creating a new charge. The guard catches the same scenario for retries that arrive after the 30-minute Stripe cache window.

Test: Stripe called exactly once despite three RECONSUME_LATER redeliveries

@SpringBootTest
@Testcontainers
class ReconsumeRedeliveryTest {

    // Uses a real RocketMQ broker in a container
    @Container
    static GenericContainer<?> rocketmq =
        new GenericContainer<>("apache/rocketmq:5.3.1")
            .withCommand("sh mqbroker -n localhost:9876 --enable-proxy")
            .withExposedPorts(9876, 8081)
            .waitingFor(Wait.forLogMessage(".*The broker.*boot success.*", 1));

    @Autowired
    private BillingConsumer billingConsumer;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9099));
        wireMock.start();
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":14900,\"status\":\"succeeded\"}")));
    }

    @Test
    void reconsumeRedelivery_stripeCalledOnceWithStableKey() {
        BillingEvent event = BillingEvent.builder()
            .subscriptionId("sub_RM001")
            .billingPeriod("2026-10")
            .amountCents(14900L)
            .stripeCustomerId("cus_test")
            .build();

        String expectedKey = sha256("sub_RM001|2026-10|14900");

        // Simulate three RECONSUME_LATER redeliveries (broker sends same event body)
        billingConsumer.accept(event);
        billingConsumer.accept(event); // first retry
        billingConsumer.accept(event); // second retry

        // Stripe received exactly one call (guard blocks subsequent calls)
        List<LoggedRequest> requests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(requests).hasSize(1);

        // Idempotency key is stable and SHA-256-derived
        assertThat(requests.get(0).getHeader("Idempotency-Key")).isEqualTo(expectedKey);

        // Exactly one BillingRecord written
        assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
            "sub_RM001", "2026-10")).isEqualTo(1);
    }

    @AfterEach
    void tearDown() { wireMock.stop(); }
}

Failure mode 2: %DLQ%billing-group message re-ingestion — on-call engineer reads DataIntegrityViolationException in DLQ headers, concludes Stripe was never called, re-publishes to billing exchange, consumer generates UUID_B committing ch_B

The setup

The billing team has an on-call runbook that includes instructions for handling dead-letter queue messages. The runbook was written when the team used RabbitMQ and describes a general procedure: if a message appears in the DLQ with a database exception in the headers, it means the message failed before any external API call (because the team historically validated DB state first), and it is safe to re-publish to the billing exchange.

After migrating to RocketMQ, the runbook was not updated. The DLQ in RocketMQ is named %DLQ%billing-group and messages land there after maxReconsumeTimes retries are exhausted. The team uses the RocketMQ Console to monitor this topic.

RocketMQ Console’s Message Detail view for DLQ messages shows:

However, the team also uses the SCS error channel to write exceptions to a separate billing-errors PostgreSQL table for audit purposes:

@ServiceActivator(inputChannel = "billing-in-0.billing-group.errors")
public void handleBillingError(ErrorMessage errorMessage) {
    MessageHandlingException cause = (MessageHandlingException) errorMessage.getPayload();
    billingErrorRepo.save(new BillingError(
        cause.getFailedMessage().getPayload().toString(),
        cause.getCause().getClass().getSimpleName(),
        cause.getCause().getMessage()
    ));
}

The billing_errors table records the exception class and message for the last failure. The on-call engineer queries this table to investigate DLQ entries.

The failure timeline

Message M1 for subscription sub_RM201 reaches the DLQ after three retries. The engineer’s investigation proceeds as follows:

The engineer’s reasoning was wrong in one critical assumption: that DataIntegrityViolationException means Stripe was not called. The actual sequence was: call Stripe (success, ch_A), then call billingRecordRepo.save() (throws DataIntegrityViolationException). The exception records the failure of the last step, not the failure of the first step. Stripe was already committed when the exception fired.

The misconception

“The exception stored in the error log for a DLQ message records the failure point of the first delivery attempt — specifically, the reason the message was rejected before it could be processed. A DataIntegrityViolationException from a DB constraint means the message hit a validation gate before any external call was made. If Stripe had been called, the failure would show a StripeException.”

This misconception conflates the exception class with the position of the failure in the processing pipeline. DataIntegrityViolationException does not indicate where in the pipeline the failure occurred — it indicates only that a DB constraint was violated. In a pipeline where Stripe is called before the DB write, the DataIntegrityViolationException fires after ch_A is already committed.

Furthermore, what the billing_errors table records is the exception from the last retry attempt, not the first. On each of the three RECONSUME_LATER redeliveries, the same exception fires at the same step: after Stripe commits. The DataIntegrityViolationException in the error log is entirely consistent with Stripe being called three times and committing three charges before the message reached the DLQ.

RocketMQ-specific context makes this particularly important to document in runbooks: unlike RabbitMQ’s x-death header (which records the reason field as rejected vs expired) or Pulsar’s DLT properties (which record the delivery count), RocketMQ’s DLQ message properties contain RECONSUMETIMES which tells you how many broker-level retries occurred — not what happened on each retry, and not whether Stripe was called on any of them.

Diagnosing this failure

Before re-ingesting any DLQ message from %DLQ%billing-group, check Stripe for existing charges under the stable idempotency key. If your consumer does not yet use a stable key, search Stripe for charges matching the subscription ID and billing period in the description field:

// Before re-ingesting: verify whether ch_A exists in Stripe
ChargeListParams params = ChargeListParams.builder()
    .setLimit(5L)
    .build();

ChargeCollection charges = stripe.charges().list(params);
boolean alreadyCharged = charges.getData().stream()
    .anyMatch(c -> c.getDescription() != null
        && c.getDescription().contains("sub_RM201")
        && c.getDescription().contains("2026-10"));

if (alreadyCharged) {
    // Do NOT re-ingest. The charge exists. Investigate the DB duplicate row instead.
    log.warn("DLQ message sub_RM201/2026-10: charge already committed in Stripe. "
           + "Fix the duplicate billing_records row, then delete the DLQ message.");
}

The root issue to fix is not the DLQ message — it is the duplicate row in billing_records that is causing every delivery attempt to throw DataIntegrityViolationException. The duplicate row is blocking the guard check from working correctly and preventing the consumer from acknowledging successfully. Fix the duplicate row first, then re-process the DLQ message (which, with the stable key, will deduplicate via Stripe’s cache).

The fix

The stable key + guard from failure mode 1 applies equally here. With the stable key, re-ingesting a DLQ message is idempotent: Stripe returns the cached ch_A response, and the guard checks for the existing BillingRecord before calling Stripe. Neither path commits ch_B.

Additionally, update your on-call runbook to replace the exception-class heuristic with a Stripe lookup:

# RUNBOOK: DLQ message in %DLQ%billing-group
#
# WRONG: "DataIntegrityViolationException = pre-Stripe failure = safe to re-ingest"
# RIGHT: Always look up the stable idempotency key in Stripe before re-ingesting.
#
# Step 1: Extract subscription_id and billing_period from the DLQ message body
# Step 2: Compute stable idempotency key:
#   key = SHA-256(subscription_id + "|" + billing_period + "|" + amount_cents)
# Step 3: Search Stripe for charges with this idempotency key:
#   GET https://api.stripe.com/v1/charges?limit=1
#   (or search by description if key is not yet stable)
# Step 4: If charge exists → fix the duplicate billing_records row,
#          then re-ingest (stable key makes it idempotent)
#         If no charge → investigate why Stripe was never called, then re-ingest

Test: DLQ re-ingestion does not commit a second Stripe charge when ch_A already exists

@SpringBootTest
@Testcontainers
class DlqReingestionTest {

    @Container
    static PostgreSQLContainer<?> postgres =
        new PostgreSQLContainer<>("postgres:16")
            .withDatabaseName("billing_test")
            .withUsername("billing")
            .withPassword("billing");

    @Autowired
    private BillingConsumer billingConsumer;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9099));
        wireMock.start();
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withBody("{\"id\":\"ch_A\",\"object\":\"charge\",\"amount\":14900,\"status\":\"succeeded\"}")));
    }

    @Test
    void dlqReingestion_stripeNotCalledWhenChArgeAlreadyRecorded() {
        BillingEvent event = BillingEvent.builder()
            .subscriptionId("sub_RM201")
            .billingPeriod("2026-10")
            .amountCents(14900L)
            .stripeCustomerId("cus_test")
            .build();

        // Simulate original delivery: Stripe commits ch_A, BillingRecord saved
        billingConsumer.accept(event);

        // Simulate DLQ re-ingestion (same event body, re-published by on-call engineer)
        billingConsumer.accept(event);

        // Stripe called exactly once — guard blocked the second call
        List<LoggedRequest> requests = wireMock.findAll(
            WireMock.postRequestedFor(WireMock.urlEqualTo("/v1/charges")));
        assertThat(requests).hasSize(1);

        // Exactly one BillingRecord written
        assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
            "sub_RM201", "2026-10")).isEqualTo(1);
        assertThat(billingRecordRepo.findBySubscriptionIdAndBillingPeriod(
            "sub_RM201", "2026-10").getStripeChargeId()).isEqualTo("ch_A");
    }

    @AfterEach
    void tearDown() { wireMock.stop(); }
}

Failure mode 3: mqadmin resetOffsetByTime to the outage window start replays already-charged messages — 1,800 pre-outage records redelivered, consumer generates UUID_B × 1,800 committing ch_B × 1,800

The setup

The billing service processes a high-volume B2B market. On a given morning, the RocketMQ nameserver experiences a brief network partition at 11:30 UTC that disconnects the billing consumer group for 6 minutes. During that window, 2,200 billing messages are published to the billing topic but not consumed. The billing consumer reconnects at 11:36 UTC and begins processing the backlog.

At 11:45 UTC, a monitoring alert fires: the consumer lag for billing-group is higher than expected. An SRE investigates and finds that the backlog is being processed, but some messages are appearing with processing errors in the application logs. The SRE decides to reset the consumer offset to ensure that no messages from the outage window are missed, using a 10-minute safety buffer:

# SRE runs this command to reset consumer offset to 10 minutes before the outage
$ mqadmin resetOffsetByTime \
    -n rocketmq-nameserver:9876 \
    -g billing-group \
    -t billing \
    -s 11:20:00 \
    -f true

# Output: reset consumer group billing-group offset for topic billing
# Queue [billing@0]: old=7,214,000  new=6,962,400  (reset to 11:20 UTC position)
# Queue [billing@1]: old=7,215,100  new=6,963,500  (reset to 11:20 UTC position)
# Queue [billing@2]: old=7,215,800  new=6,964,200  (reset to 11:20 UTC position)
# Queue [billing@3]: old=7,214,600  new=6,963,800  (reset to 11:20 UTC position)
# Offset reset successfully

The SRE’s intent: reset to 10 minutes before the outage to ensure the full 11:30–11:36 UTC window is covered. The safety buffer is standard incident runbook practice.

The failure timeline

Total damage: 1,800 customers double-charged. The 2,200 outage-window messages are processed correctly. The offset reset did exactly what was needed for the outage window and also exactly what was not needed for the pre-outage window.

The misconception

“mqadmin resetOffsetByTime only affects unprocessed messages. RocketMQ tracks which message IDs were successfully consumed and acknowledged by a consumer group in a persistent consumed-set. When the offset is reset to an earlier timestamp, the broker will replay messages from that timestamp, but it will compare each message’s ID against the consumed-set for the consumer group and skip any messages that were previously acknowledged. The reset is safe to apply with a generous buffer because already-processed messages will not be redelivered.”

RocketMQ does not maintain a per-message-ID consumed-set. Consumer group progress is tracked as a single numeric offset per message queue (a queue is a partition in Kafka terminology). The offset is the position of the next message to consume. When resetOffsetByTime overwrites the stored offset to an earlier position, the consumer group will receive every message at or after that position, regardless of whether it was previously acknowledged.

This is the same semantics as Kafka’s --reset-offsets --to-datetime, Pulsar’s resetCursor, and RabbitMQ’s message replay via Shovel. None of these tools maintain a skip-list of previously-acknowledged message IDs. They all operate on the offset or cursor position only.

The -f true flag in the SRE’s command tells mqadmin to force the reset even if the consumer group is currently active and consuming. Without -f true, the reset requires the consumer group to be offline. The fact that the reset succeeds while the consumer is running means the 1,800 already-processed messages enter the consumer’s queue immediately after the reset, interleaved with the new messages from the outage window.

The RocketMQ offset model: what resetOffsetByTime actually does

RocketMQ stores consumer group offsets in the broker’s offset store. For each (consumerGroup, topic, queueId) triple, the offset store holds a single 64-bit integer: the position of the next unconsumed message. When the consumer sends a ConsumeConcurrentlyStatus.CONSUME_SUCCESS acknowledgement for a batch, the broker advances the stored offset to the position after the last acknowledged message in the batch.

mqadmin resetOffsetByTime queries the broker’s message index for each queue to find the position of the first message published at or after the specified timestamp. It then writes that position as the new stored offset for the (consumerGroup, topic, queueId) triple, overwriting whatever the previous offset was. The consumer will next fetch starting from the new position, and will receive all messages at or after that position.

# What resetOffsetByTime does to the offset store:
#
# Before reset:
#   billing-group / billing / queue-0 : offset = 7,214,000 (11:45 UTC position)
#
# resetOffsetByTime -s 11:20:00
#   Broker looks up: "what offset corresponds to messages published at 11:20 UTC?"
#   Answer: 6,962,400
#
# After reset:
#   billing-group / billing / queue-0 : offset = 6,962,400 (11:20 UTC position)
#
# Next consumer fetch from queue-0 will start at offset 6,962,400.
# Messages 6,962,400 through 7,213,999 will all be redelivered.
# That range includes:
#   - 11:20–11:30 UTC messages: 1,800 messages (already charged)
#   - 11:30–11:36 UTC messages: 2,200 messages (not yet charged)
#   - 11:36–11:45 UTC messages: any messages processed since reconnect

The fix

The stable key + guard makes resetOffsetByTime replay idempotent. When the consumer processes the 1,800 already-charged messages, the guard checks for an existing BillingRecord per subscription per billing period. All 1,800 find an existing record and return early without calling Stripe. The 2,200 outage-window messages are processed normally.

Additionally, update your offset-reset runbook to use the exact outage timestamp rather than a safety buffer. The safety buffer is a Kafka-era heuristic that assumes offset resets are idempotent for already-processed messages — which is only true when your consumer has idempotency built in.

# RUNBOOK: Offset reset for RocketMQ consumer group after an outage
#
# WRONG (before idempotency fix):
# $ mqadmin resetOffsetByTime -s <outage_start - 10_minutes>
# Risk: replays all messages in the safety buffer window — any message
#       that was processed in that window will be redelivered and may
#       produce a duplicate Stripe charge if the consumer uses UUID.randomUUID().
#
# CORRECT (with stable key + guard):
# $ mqadmin resetOffsetByTime -s <exact_outage_start_timestamp>
# Safety: the stable key + guard makes any replay idempotent.
#         The 10-minute buffer adds no risk because already-processed messages
#         hit the BillingRecord guard and return early.
#
# Verify before reset:
# 1. Confirm billing-group.billing-period for the reset window matches
#    expected charge count in Stripe dashboard
# 2. Run: mqadmin queryConsumerStatus -g billing-group -t billing
#    to confirm the consumer group is healthy before resetting
#
# Example:
$ mqadmin resetOffsetByTime \
    -n rocketmq-nameserver:9876 \
    -g billing-group \
    -t billing \
    -s 11:30:00 \   # exact outage start, not 11:20
    -f true

Test: resetOffsetByTime replay does not commit duplicate Stripe charges

@SpringBootTest
@Testcontainers
class OffsetResetReplayTest {

    @Container
    static PostgreSQLContainer<?> postgres =
        new PostgreSQLContainer<>("postgres:16")
            .withDatabaseName("billing_test")
            .withUsername("billing")
            .withPassword("billing");

    @Autowired
    private BillingConsumer billingConsumer;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    private WireMockServer wireMock;

    @BeforeEach
    void setUp() {
        wireMock = new WireMockServer(WireMockConfiguration.wireMockConfig().port(9099));
        wireMock.start();
        // Stripe returns unique charge IDs per idempotency key
        wireMock.stubFor(WireMock.post(WireMock.urlEqualTo("/v1/charges"))
            .willReturn(WireMock.aResponse()
                .withStatus(200)
                .withHeader("Content-Type", "application/json")
                .withTransformerParameter("responseTemplatingEnabled", true)
                .withBody("{\"id\":\"ch_{{randomValue length=16 type='ALPHANUMERIC'}}\","
                        + "\"object\":\"charge\",\"amount\":14900,\"status\":\"succeeded\"}")));
    }

    @Test
    void offsetResetReplay_alreadyChargedMessagesSkipped() {
        // Simulate 5 messages processed before the outage — all already charged
        List<BillingEvent> preOutageEvents = IntStream.range(0, 5)
            .mapToObj(i -> BillingEvent.builder()
                .subscriptionId("sub_pre_" + i)
                .billingPeriod("2026-10")
                .amountCents(14900L)
                .stripeCustomerId("cus_test_" + i)
                .build())
            .toList();

        preOutageEvents.forEach(billingConsumer::accept);
        assertThat(wireMock.getAllServeEvents()).hasSize(5);

        // Simulate resetOffsetByTime — same messages redelivered
        preOutageEvents.forEach(billingConsumer::accept);

        // Stripe still only called 5 times — guard blocked all 5 replays
        assertThat(wireMock.getAllServeEvents()).hasSize(5);

        // 5 BillingRecords, one per subscription
        for (int i = 0; i < 5; i++) {
            assertThat(billingRecordRepo.countBySubscriptionIdAndBillingPeriod(
                "sub_pre_" + i, "2026-10")).isEqualTo(1);
        }
    }

    @AfterEach
    void tearDown() { wireMock.stop(); }
}

Eight-item diagnostic checklist for SCS RocketMQ Binder Stripe billing consumers

Run this checklist against any billing consumer using the Spring Cloud Stream RocketMQ Binder before deploying to production.

  1. Idempotency key source. Every call to stripe.charges().create(), stripe.paymentIntents().create(), or any Stripe write API in the consumer must use an idempotency key derived from stable payload fields (subscription ID, billing period, amount), not from UUID.randomUUID(). UUID.randomUUID() produces a different value on every invocation; in RocketMQ, every RECONSUME_LATER redelivery and every resetOffsetByTime replay is a new invocation.
  2. maxReconsumeTimes setting and Stripe call count. Confirm the value of spring.cloud.stream.rocketmq.bindings.<binding>.consumer.maxReconsumeTimes. The default is -1 (16 retries). With UUID.randomUUID() and no guard, the maximum number of Stripe charges per message is maxReconsumeTimes + 1. With the stable key + guard, the retry count is irrelevant.
  3. Retry delay level awareness. Know your RocketMQ default delay levels: 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h. These determine how quickly duplicate charges appear in Stripe’s dashboard after the original charge. A charge that duplicates 10 seconds after the original is a RocketMQ first-retry signature.
  4. DLQ runbook accuracy. Verify that your on-call runbook for %DLQ%billing-group does not contain exception-class heuristics for determining Stripe call status. Replace with: compute the stable idempotency key from the DLQ message payload, look up the key in Stripe, decide whether to re-ingest based on the lookup result.
  5. Guard-before-Stripe check. Confirm that the billing consumer checks for an existing BillingRecord by the stable idempotency key before every Stripe call. This guard is the defense against retries arriving after Stripe’s 30-minute idempotency cache has expired and against replays from resetOffsetByTime that go back more than 30 minutes.
  6. Offset reset runbook timestamp precision. If your SRE runbook for outage recovery includes a step for resetting the consumer group offset, replace “outage start minus safety buffer” with “exact outage start timestamp”. A generous buffer only adds risk if the consumer lacks idempotency; with idempotency built in, the exact timestamp is strictly more conservative and no less safe.
  7. Message model configuration. Confirm spring.cloud.stream.rocketmq.bindings.<binding>.consumer.messageModel is set to CLUSTERING (the default) and not BROADCASTING. BROADCASTING mode delivers every message to every consumer instance in the group, which with a 3-pod billing service means 3 Stripe charges per message regardless of idempotency key (because all three pods receive the same message independently and call Stripe with the same key — Stripe’s idempotency cache deduplicates concurrent calls with the same key, but only within a 30-minute window and only if the calls arrive close together).
  8. Orderly vs concurrent consumption mode. If spring.cloud.stream.rocketmq.bindings.<binding>.consumer.orderly=true, the binder uses MessageListenerOrderly instead of MessageListenerConcurrently. In orderly mode, failed messages return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT, which keeps the message at the head of the queue and retries without the %RETRY% topic indirection. The retry semantics are similar (no persistent state between retries, UUID.randomUUID() produces a new key on each attempt), but the timing and backoff behavior differ. Confirm your idempotency implementation works in both modes.

RocketMQ-specific Stripe double-charge modes vs Kafka and RabbitMQ

Aspect RocketMQ (SCS binder) Kafka (SCS binder) RabbitMQ (SCS binder)
Retry mechanism Broker routes to %RETRY%<group> topic after RECONSUME_LATER Application-side retry within same listener invocation; broker not involved Application-side retry (SimpleMessageListenerContainer); message requeued in original queue after exhaustion
Retry state persistence None between retries; each retry is a new broker delivery from %RETRY% Application-side; same ConsumerRecord object, in-process retry None between retries; same message body but fresh listener invocation after requeue
DLQ / DLT name %DLQ%<consumerGroup> <topic>.DLT (configurable) <exchange>.dlq (configurable)
DLQ failure evidence RECONSUMETIMES property (delivery count only); exception not stored in message properties by default Exception type + stack trace in DLT headers via DeadLetterPublishingRecoverer x-death header with reason, count, exchange, routing-keys
Offset reset tool mqadmin resetOffsetByTime kafka-consumer-groups.sh --reset-offsets --to-datetime No direct offset concept; shovel / re-queue from management UI
Offset reset semantics Overwrites per-queue offset; all messages at or after new position redelivered Overwrites per-partition committed offset; same semantics N/A; message-centric model
Skip already-consumed messages on reset? No. No per-message-ID consumed-set. No. No per-message-ID consumed-set. N/A

Summary: three SCS RocketMQ Binder Stripe failure modes at a glance

Failure mode Trigger Why UUID_B Fix
RECONSUME_LATER retry DB write throws after ch_A commits; binder returns RECONSUME_LATER; broker routes to %RETRY%billing-group Retry is a new broker delivery with no in-memory state; UUID.randomUUID() produces UUID_B on fresh invocation Stable SHA-256 key (same on every retry); BillingRecord guard before Stripe call
%DLQ%billing-group re-ingestion On-call engineer reads DataIntegrityViolationException in error log, concludes Stripe was never called, re-publishes DLQ message Re-ingested message is a new consumer invocation; UUID.randomUUID() produces UUID_B; DLQ headers do not confirm whether Stripe was called on prior attempts Stable key + guard makes re-ingestion idempotent; runbook must check Stripe before re-ingesting
mqadmin resetOffsetByTime replay SRE resets consumer group offset to timestamp before outage with safety buffer; 1,800 pre-outage already-charged messages replayed Replayed messages are new deliveries; UUID.randomUUID() produces UUID_B × 1,800; RocketMQ has no per-message consumed-set that survives offset resets Stable key + guard makes replay idempotent; runbook should use exact outage timestamp, not buffer

Stop duplicate Stripe charges from RocketMQ retries

Scoped vault keys, per-vendor spend caps, allowed-endpoint allowlists, and a queryable audit log for every Stripe call your consumer makes. Detect RECONSUME_LATER retry duplicates and resetOffsetByTime replay charges before the card dispute arrives.