Spring Integration Azure Event Hubs Stripe Integration: How `ExecutorChannel` Commits the RECORD Checkpoint When the Message Is Queued Not When Processing Completes Enabling `ch_B` on `errorChannel` Retry, How `RequestHandlerRetryAdvice` Calls `UUID.randomUUID()` Fresh on Each Retry Attempt Generating `UUID_B` After a Stripe 200 Lost In Transit Commits `ch_B`, and How `EventHubsInboundChannelAdapter` Restart Creates a New `EventProcessorClient` That Claims the Partition Before the Old Client’s 3DS Call Completes Generating `UUID_B` Committing `ch_B`

By Keybrake · October 29, 2026 · 9,100 words

Spring Integration and Spring Cloud Stream share the same Azure Event Hubs Java SDK underneath, but their checkpoint and retry architectures differ in ways that create unique Stripe double-charge scenarios. The three failure modes below require understanding Spring Integration’s channel type semantics, advice chain execution order, and MessageProducerSupport lifecycle — none of which appear in the SCS AEH Binder, the raw EventProcessorClient, or the Spring Cloud Stream Azure Service Bus Binder documentation.

Spring Integration Azure Event Hubs Architecture

The spring-cloud-azure-starter-integration-eventhubs artifact provides three primary components: EventHubsInboundChannelAdapter (reads events from Event Hubs via EventHubsMessageListenerContainer), EventHubsMessageHandler (writes events via EventHubsTemplate), and EventHubsMessageListenerContainer (wraps the Azure SDK’s EventProcessorClient with Spring Integration lifecycle management).

The EventHubsInboundChannelAdapter extends MessageProducerSupport, which extends AbstractEndpoint — giving it a Spring-managed start/stop lifecycle hooked to ApplicationContext events. This lifecycle management is the source of the third failure mode below.

Checkpoint behavior in Spring Integration AEH is configured on the EventHubsContainerProperties that the EventHubsMessageListenerContainer uses:

@Bean
public EventHubsMessageListenerContainer listenerContainer(
        EventHubsProcessorFactory processorFactory) {
    EventHubsContainerProperties props = new EventHubsContainerProperties();
    props.setEventHubName("billing-events");
    props.setConsumerGroup("billing-consumer");
    props.setCheckpointConfig(new CheckpointConfig(CheckpointMode.RECORD));
    return new EventHubsMessageListenerContainer(processorFactory, props);
}

@Bean
public EventHubsInboundChannelAdapter adapter(
        EventHubsMessageListenerContainer container) {
    EventHubsInboundChannelAdapter adapter =
        new EventHubsInboundChannelAdapter(container);
    adapter.setOutputChannel(billingChannel());
    return adapter;
}

The supported checkpoint modes are:

Mode	When checkpoint fires	Who triggers it
`RECORD`	After each message is delivered to the output channel and the `send()` call returns	Container, automatically
`BATCH`	After each batch of messages is delivered	Container, automatically
`MANUAL`	When the application calls `Checkpointer.updateCheckpoint()`	Application code via `AzureHeaders.CHECKPOINTER`
`PARTITION_COUNT`	After N messages per partition	Container, automatically
`TIME`	On a fixed wall-clock interval	Container, automatically
`COUNT_TIME`	Whichever of COUNT or TIME fires first	Container, automatically

The critical detail — and the one that creates the first failure mode — is the phrase “after the send() call returns.” What this means in practice depends on which type of MessageChannel the adapter’s outputChannel is wired to.

How This Differs from Spring Cloud Stream AEH Binder

In Spring Cloud Stream’s Azure Event Hubs Binder, the consumer function (Consumer<EventData> or Consumer<List<EventData>>) is always executed synchronously on the binder’s internal thread before the checkpoint fires. The SCS binder does not expose the concept of a MessageChannel type to the application developer — the binder manages the channel topology internally. Retry behavior is controlled by spring.cloud.stream.bindings.billing-in-0.consumer.max-attempts at the binder level, not by Spring Integration advice chains.

In Spring Integration, the developer explicitly wires the EventHubsInboundChannelAdapter to a MessageChannel of their choice — DirectChannel, ExecutorChannel, QueueChannel, PriorityChannel, RendezvousChannel, or PublishSubscribeChannel. Each channel type has different send() semantics, and those semantics directly affect when RECORD checkpoint fires relative to when the downstream ServiceActivator processes the message.

Channel type	`send()` blocks until?	RECORD checkpoint fires	SI retry on SA failure?
`DirectChannel`	ServiceActivator returns or throws	After SA completes (safe)	Exception propagates to adapter
`ExecutorChannel`	Task submitted to executor	Before SA executes (unsafe)	Error channel, new UUID
`QueueChannel`	Message queued (in-memory)	Before SA executes (unsafe)	Error channel, new UUID
`PriorityChannel`	Message queued (priority sorted)	Before SA executes (unsafe)	Error channel, new UUID
`PublishSubscribeChannel`	All subscribers (sync) or submitted (async)	Depends on executor config	Depends on executor config

Only DirectChannel is safe for Stripe billing flows with automatic checkpoint modes. Every other channel type creates a window in which the checkpoint commits before the Stripe call happens.

Failure Mode 1: `ExecutorChannel` Commits the RECORD Checkpoint at Queue Time, Not at Processing Completion Time

The Setup

The billing service receives high-volume usage events from Azure Event Hubs and calls Stripe to bill customers. To avoid blocking the Event Hubs listener thread (which would stall partition processing), the team routes events through an ExecutorChannel backed by a fixed thread pool, allowing parallel Stripe API calls across multiple threads.

@Configuration
public class BillingIntegrationConfig {

    @Bean
    public MessageChannel billingChannel() {
        // Thread pool for parallel Stripe API calls
        return new ExecutorChannel(
            Executors.newFixedThreadPool(8, new ThreadFactoryBuilder()
                .setNameFormat("billing-executor-%d")
                .build())
        );
    }

    @Bean
    public EventHubsMessageListenerContainer listenerContainer(
            EventHubsProcessorFactory processorFactory) {
        EventHubsContainerProperties props = new EventHubsContainerProperties();
        props.setEventHubName("billing-events");
        props.setConsumerGroup("$Default");
        props.setCheckpointConfig(new CheckpointConfig(CheckpointMode.RECORD));
        return new EventHubsMessageListenerContainer(processorFactory, props);
    }

    @Bean
    public EventHubsInboundChannelAdapter billingAdapter(
            EventHubsMessageListenerContainer container) {
        EventHubsInboundChannelAdapter adapter =
            new EventHubsInboundChannelAdapter(container);
        adapter.setOutputChannel(billingChannel()); // ExecutorChannel!
        return adapter;
    }
}

@MessageEndpoint
public class BillingServiceActivator {

    @Autowired private StripeClient stripeClient;
    @Autowired private BillingRecordRepository billingRecordRepo;

    @ServiceActivator(inputChannel = "billingChannel")
    public void processBilling(Message<EventData> message) {
        BillingEvent event = deserializeEvent(message.getPayload());

        // BUG: fresh UUID on every invocation
        String idempotencyKey = UUID.randomUUID().toString();

        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getStripeCustomerId())
            .setDescription("Usage " + event.getAccountId()
                + " / " + event.getBillingPeriod())
            .build();

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

        billingRecordRepo.save(new BillingRecord(
            event.getAccountId(),
            event.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    }
}

The Failure Sequence

Event E1 (offset 4,201, partition 3) arrives at the EventHubsInboundChannelAdapter. The container calls adapter.onMessage(eventContext). The adapter converts the Azure EventData to a Spring Message<EventData> and calls billingChannel.send(message).

Here is the critical divergence. ExecutorChannel.send() submits the task to the thread pool queue and returns true immediately. It does NOT wait for the executor thread to pick up the task or for the ServiceActivator to process it. The call stack returns to the container’s onMessage() handler, which sees that send() returned without throwing, and fires the RECORD checkpoint:

// Inside EventHubsMessageListenerContainer (simplified)
void onMessage(EventContext eventContext) {
    Message<EventData> message = convertToMessage(eventContext);
    outputChannel.send(message);   // returns true immediately for ExecutorChannel
    // RECORD checkpoint fires HERE, before executor thread runs
    if (checkpointMode == RECORD) {
        eventContext.updateCheckpoint();  // offset 4201 committed to Blob Storage
    }
}

Azure Blob Storage now records: partition 3 checkpoint = offset 4,201. The event is considered processed from Event Hubs’ perspective.

Meanwhile, on an executor thread: BillingServiceActivator.processBilling() begins. UUID.randomUUID() generates UUID_A = 3f2a9.... stripe.charges.create(UUID_A) is called. Stripe receives the request at t=0. At t=480ms, Stripe commits the charge on the server side (ch_A). At t=500ms, the OkHttp client’s read timeout fires (configured at 500ms to keep Stripe p99 latency acceptable). StripeConnectException("Read timeout") is thrown.

The executor thread’s uncaught exception handler sends an ErrorMessage wrapping the original message to the error channel. The error channel’s subscriber — a ServiceActivatingHandler with retry logic — picks it up. The retry handler calls BillingServiceActivator.processBilling() with the original Message<EventData> object.

UUID.randomUUID() generates UUID_B = 8c71d.... stripe.charges.create(UUID_B) is called. Stripe has no record of UUID_B (different from UUID_A). Stripe commits ch_B.

Both ch_A and ch_B now appear on the customer’s Stripe account for the same billing period.

The Developer Misconception

“RECORD checkpoint mode checkpoints after the ServiceActivator processes the message. The adapter waits for the downstream processing to complete before committing the offset.”

This is true for DirectChannel (synchronous, exception propagates back to send()) but false for any asynchronous channel type. The RECORD checkpoint fires when outputChannel.send() returns — not when the downstream handler returns. For ExecutorChannel, these two events are separated by the entire duration of the executor thread’s work.

Spring Integration’s official documentation states: “For ExecutorChannel, the sender’s thread is separate from the subscriber’s thread. Messages are dispatched to subscribers using the provided TaskExecutor, and the calling thread is released immediately after dispatching.” The implication for checkpoint-triggering adapters is rarely called out explicitly.

The Fix: Use `DirectChannel` for Billing Flows, or Switch to `MANUAL` Checkpoint

The fundamental fix is to use DirectChannel as the output channel of the EventHubsInboundChannelAdapter for any billing flow that requires at-least-once delivery semantics with automatic checkpoint modes:

@Bean
public MessageChannel billingChannel() {
    // DirectChannel: send() blocks until ServiceActivator returns or throws
    // Exception propagates back to adapter, preventing RECORD checkpoint
    return new DirectChannel();
}

With DirectChannel, if the ServiceActivator throws, the exception propagates synchronously through send() back to the container’s onMessage() handler. The updateCheckpoint() call never fires. On pod restart, the EventProcessorClient reads from the last committed checkpoint (the offset before E1), and E1 is redelivered.

If parallel processing is required for throughput, the parallelism should be achieved by scaling the number of partitions and pod replicas, not by routing through an async channel. Each pod’s adapter processes its assigned partitions sequentially on a DirectChannel, and throughput scales with partition count.

If ExecutorChannel is genuinely needed (e.g., for non-billing downstream flows on the same adapter), switch to CheckpointMode.MANUAL and checkpoint only after the guard write inside the ServiceActivator:

// Container config
props.setCheckpointConfig(new CheckpointConfig(CheckpointMode.MANUAL));

@ServiceActivator(inputChannel = "billingChannel")
public void processBilling(Message<EventData> message) {
    Checkpointer checkpointer = message.getHeaders()
        .get(AzureHeaders.CHECKPOINTER, Checkpointer.class);
    BillingEvent event = deserializeEvent(message.getPayload());

    String idempotencyKey = Hashing.sha256()
        .hashString(event.getAccountId() + "|" + event.getBillingPeriod()
            + "|" + event.getAmountCents() + "|" + event.getUnitsConsumed(),
            StandardCharsets.UTF_8).toString();

    // Guard: skip if already billed
    if (billingRecordRepo.existsByAccountIdAndBillingPeriod(
            event.getAccountId(), event.getBillingPeriod())) {
        checkpointer.updateCheckpoint(message).block();
        return;
    }

    Charge charge = stripeClient.charges().create(
        ChargeCreateParams.builder()
            .setAmount(event.getAmountCents())
            .setCurrency("usd")
            .setCustomer(event.getStripeCustomerId())
            .build(),
        RequestOptions.builder().setIdempotencyKey(idempotencyKey).build()
    );

    // Write guard FIRST — only checkpoint after guard is durable
    billingRecordRepo.save(new BillingRecord(
        event.getAccountId(), event.getBillingPeriod(),
        charge.getId(), idempotencyKey
    ));

    checkpointer.updateCheckpoint(message).block(); // offset committed last
}

With MANUAL checkpoint mode and ExecutorChannel, the checkpoint is committed only when the ServiceActivator explicitly calls checkpointer.updateCheckpoint().block() — after the guard record is durable. If the Stripe call or the guard write fails, the checkpoint is never committed, and the event is redelivered on pod restart.

Test: Verifying `ExecutorChannel` Checkpoint Timing

@SpringBootTest
@Testcontainers
class ExecutorChannelCheckpointTest {

    @Container
    static AzuriteContainer azurite = new AzuriteContainer("mcr.microsoft.com/azure-storage/azurite:3.29.0");

    @MockBean
    private StripeClient stripeClient;

    @Autowired
    private EventHubsInboundChannelAdapter billingAdapter;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    @Test
    void executorChannel_checkpointFiresBeforeServiceActivator_allowsDoubleCharge() throws Exception {
        // Arrange: Stripe returns HTTP 200 on first call, 200 on second call
        // Both succeed → double charge
        AtomicInteger callCount = new AtomicInteger(0);
        when(stripeClient.charges().create(any(), any()))
            .thenAnswer(invocation -> {
                int n = callCount.incrementAndGet();
                Charge charge = mock(Charge.class);
                when(charge.getId()).thenReturn("ch_" + (n == 1 ? "A" : "B"));
                return charge;
            });

        // Act: inject a billing event that triggers Stripe timeout on first attempt
        // then succeeds on error channel retry
        // (Stripe mock simulates: first call takes 600ms, second call takes 50ms)
        when(stripeClient.charges().create(argThat(p ->
                callCount.get() == 0), any()))
            .thenAnswer(inv -> { Thread.sleep(600); throw new StripeException("timeout", null, null, 0, null){}; });

        EventData eventData = buildBillingEventData("acc-1", "2026-10", 9900L, 99L);
        adapter.onMessage(buildEventContext(eventData, 4201L, 3));

        // Wait for error channel retry to complete
        Thread.sleep(3000);

        // Assert: two BillingRecords saved = double charge
        List<BillingRecord> records = billingRecordRepo
            .findByAccountIdAndBillingPeriod("acc-1", "2026-10");
        assertThat(records).hasSize(2);
        assertThat(records).extracting(BillingRecord::getChargeId)
            .containsExactlyInAnyOrder("ch_A", "ch_B");
    }

    @Test
    void directChannel_exceptionPreventsCheckpoint_noDoubleCharge() throws Exception {
        // With DirectChannel, Stripe timeout propagates back to adapter
        // Checkpoint not committed → redelivery → same idempotency key → safe

        when(stripeClient.charges().create(any(), any()))
            .thenThrow(new StripeException("timeout", null, null, 0, null){})
            .thenAnswer(inv -> {
                Charge charge = mock(Charge.class);
                when(charge.getId()).thenReturn("ch_A");
                return charge;
            });

        EventData eventData = buildBillingEventData("acc-2", "2026-10", 4900L, 49L);
        adapter.onMessage(buildEventContext(eventData, 4202L, 3));
        // Redelivered after checkpoint not committed
        adapter.onMessage(buildEventContext(eventData, 4202L, 3));

        List<BillingRecord> records = billingRecordRepo
            .findByAccountIdAndBillingPeriod("acc-2", "2026-10");
        // Only one charge committed on redelivery (guard skips first if ch_A was
        // committed during the failed first attempt — stable key ensures idempotency)
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getChargeId()).isEqualTo("ch_A");
    }
}

Failure Mode 2: `RequestHandlerRetryAdvice` Re-invokes the Entire Service Activator Including `UUID.randomUUID()` on Each Retry Attempt

The Setup

The billing team correctly uses DirectChannel after learning about the ExecutorChannel checkpoint timing issue. They add a RequestHandlerRetryAdvice to the ServiceActivator to automatically retry transient failures — Stripe rate limits, temporary DB unavailability, and network blips. The retry advice is configured with exponential backoff up to 3 attempts.

@Configuration
public class BillingIntegrationConfig {

    @Bean
    public MessageChannel billingChannel() {
        return new DirectChannel(); // Correct: synchronous
    }

    @Bean
    public RequestHandlerRetryAdvice billingRetryAdvice() {
        RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
        advice.setRetryTemplate(RetryTemplate.builder()
            .maxAttempts(3)
            .exponentialBackoff(500, 2.0, 4000)
            .retryOn(StripeException.class)
            .retryOn(DataAccessException.class)
            .build());
        return advice;
    }

    @Bean
    public IntegrationFlow billingFlow(
            EventHubsInboundChannelAdapter adapter,
            BillingServiceActivator billingService) {
        return IntegrationFlow
            .from(adapter)
            .handle(billingService, "processBilling",
                e -> e.advice(billingRetryAdvice()))
            .get();
    }
}

@MessageEndpoint
public class BillingServiceActivator {

    @ServiceActivator
    public void processBilling(Message<EventData> message) {
        BillingEvent event = deserializeEvent(message.getPayload());

        // BUG: UUID.randomUUID() called inside the method body —
        // RequestHandlerRetryAdvice re-invokes this method on every attempt
        String idempotencyKey = UUID.randomUUID().toString();

        Charge charge = stripeClient.charges().create(
            ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Usage " + event.getAccountId()
                    + " / " + event.getBillingPeriod())
                .build(),
            RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build()
        );

        billingRecordRepo.save(new BillingRecord(
            event.getAccountId(),
            event.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    }
}

How `RequestHandlerRetryAdvice` Works

RequestHandlerRetryAdvice is a Spring AOP advice that wraps a MessageHandler’s handleRequestMessage() method. When the handler throws a retryable exception, the advice’s RetryTemplate calls handler.handleRequestMessage(message) again — passing the same Message object from the channel. The advice does not inspect or modify the message payload or headers between retries. The handler’s method body is executed fresh from the top on each attempt.

This means every line of processBilling() — including UUID.randomUUID() — is executed on every retry attempt. The returned UUID is different on every attempt because randomUUID() is explicitly designed to return a new value every time.

The Failure Sequence

Event E2 (offset 9,782, partition 11) arrives. The EventHubsInboundChannelAdapter sends it to the DirectChannel. The ServiceActivator is invoked:

Attempt 1, t=0ms: UUID.randomUUID() → UUID_A = 4a71f.... stripe.charges.create(params, idempotencyKey=UUID_A) called. The HTTP request reaches Stripe’s servers at t=80ms. Stripe validates the request, creates the PaymentIntent, confirms the charge, and commits ch_A on the server side at t=430ms. Stripe’s API server begins writing the HTTP 200 response. At t=480ms, a network switch in the cloud provider’s backbone drops the TCP connection between the JVM and Stripe’s API endpoint. The Stripe Java SDK’s OkHttpClient receives a SocketException("Connection reset"). The SDK wraps this in a StripeConnectException and throws it.
RetryAdvice: StripeConnectException is retryable. Waiting 500ms (first backoff interval).
Attempt 2, t=980ms: UUID.randomUUID() → UUID_B = 9c22e.... stripe.charges.create(params, idempotencyKey=UUID_B) called. Stripe has no record of UUID_B (it only knows about UUID_A, which is now associated with ch_A). Stripe processes the request as a new charge. ch_B committed at t=1,450ms. HTTP 200 arrives successfully.

The ServiceActivator saves a BillingRecord for ch_B. No exception is thrown. The RequestHandlerRetryAdvice considers the operation successful. The RECORD checkpoint fires. Partition 11 offset 9,782 is committed to Blob Storage.

The customer’s Stripe account shows ch_A (from attempt 1, committed but response lost) and ch_B (from attempt 2, committed and acknowledged). The billing team’s database shows one BillingRecord with ch_B. ch_A is untracked, orphaned on Stripe.

The Developer Misconception

“RequestHandlerRetryAdvice retries with the same Message object — the same headers including any idempotency key that was previously stored in a header. Only the part of the handler that threw is re-executed, not the idempotency key generation.”

The advice does pass the same Message object. But it re-invokes the entire handler method from the beginning — including any UUID generation that happens inside the method body. If the idempotency key is generated as a header or attribute of the incoming Message before the service activator is called (e.g., by an upstream HeaderEnricher), then the key is stable across retries. If the key is generated inside the service activator using UUID.randomUUID(), it is different on each attempt.

This is in contrast to how Spring Cloud Stream’s retry works: the SCS binder’s max-attempts also re-invokes the consumer function from scratch, with the same behavior. The difference is that Spring Integration developers using RequestHandlerRetryAdvice often assume it is more targeted than it actually is — that it somehow “knows” which part of the handler failed and only retries that part.

The Fix: Stable Content-Hash Key + `BillingRecord` Guard

@MessageEndpoint
public class BillingServiceActivator {

    @ServiceActivator
    public void processBilling(Message<EventData> message) {
        BillingEvent event = deserializeEvent(message.getPayload());

        // Stable key: same value on every retry for the same event
        String idempotencyKey = Hashing.sha256()
            .hashString(
                event.getAccountId() + "|"
                + event.getBillingPeriod() + "|"
                + event.getAmountCents() + "|"
                + event.getUnitsConsumed(),
                StandardCharsets.UTF_8
            ).toString();

        // Guard: if already billed (ch_A may have committed even if we got SocketException)
        if (billingRecordRepo.existsByAccountIdAndBillingPeriod(
                event.getAccountId(), event.getBillingPeriod())) {
            log.info("billing already recorded for {}/{}, skipping Stripe",
                event.getAccountId(), event.getBillingPeriod());
            return;
        }

        Charge charge = stripeClient.charges().create(
            ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .setDescription("Usage " + event.getAccountId()
                    + " / " + event.getBillingPeriod())
                .build(),
            RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build()
        );

        billingRecordRepo.save(new BillingRecord(
            event.getAccountId(),
            event.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    }
}

The stable content-hash key ensures that if the Stripe call succeeded but the response was lost, the retry sends the same key. Stripe responds to the second request with the same Charge object (the idempotent response for that key, which is cached by Stripe for 24 hours). The guard additionally handles the case where the response was lost AND the retry comes in with a different key — if ch_A was committed and the billing record was saved before the exception (e.g., failure was in a post-Stripe step), the guard will detect the existing record and skip the retry entirely.

Alternative: Generate the Key Upstream in a `HeaderEnricher`

If the billing service activator must remain stateless (no database access in the handler), the idempotency key can be generated by an upstream HeaderEnricher transformer. Because the HeaderEnricher runs before the RequestHandlerRetryAdvice’s retry scope, the enriched header is set once and the same Message object (with the same headers) is passed to the service activator on every retry attempt:

@Bean
public IntegrationFlow billingFlow(
        EventHubsInboundChannelAdapter adapter,
        BillingServiceActivator billingService) {
    return IntegrationFlow
        .from(adapter)
        // HeaderEnricher runs once — before retry scope
        .enrichHeaders(h -> h.headerFunction(
            "stripe-idempotency-key",
            message -> {
                BillingEvent event = deserializeEvent(
                    ((EventData) message.getPayload()));
                return Hashing.sha256()
                    .hashString(
                        event.getAccountId() + "|"
                        + event.getBillingPeriod() + "|"
                        + event.getAmountCents() + "|"
                        + event.getUnitsConsumed(),
                        StandardCharsets.UTF_8
                    ).toString();
            }
        ))
        // RetryAdvice wraps only the service activator
        .handle(billingService, "processBilling",
            e -> e.advice(billingRetryAdvice()))
        .get();
}

@MessageEndpoint
public class BillingServiceActivator {

    @ServiceActivator
    public void processBilling(Message<EventData> message) {
        // Key from header — stable across all retry attempts of this Message
        String idempotencyKey = (String) message.getHeaders()
            .get("stripe-idempotency-key");

        stripeClient.charges().create(
            buildParams(message),
            RequestOptions.builder().setIdempotencyKey(idempotencyKey).build()
        );
        // ...
    }
}

The HeaderEnricher transformer runs exactly once per message, upstream of the retry scope. The RequestHandlerRetryAdvice’s retry loop re-calls billingService.processBilling(message) with the same Message object — including the same stripe-idempotency-key header set by the enricher. Stripe returns the same Charge response for the same key.

Comparison: Spring Integration Advice Chain vs. SCS AEH Binder Retry

Property	SI `RequestHandlerRetryAdvice`	SCS AEH Binder `max-attempts`
Scope	Wraps one `ServiceActivator` (or `MessageHandler`)	Wraps the entire consumer `Function` invocation
What is retried	The `handleRequestMessage()` method body	The full `Consumer<T>.accept()` call
UUID.randomUUID() behavior	New UUID on each retry (method re-invoked from start)	New UUID on each retry (function re-invoked from start)
Message object on retry	Same `Message<T>` object, same headers	Same converted payload, same headers
Retry configuration	`RetryTemplate` on the advice bean	`spring.cloud.stream.bindings.*.consumer.max-attempts`
Error handling after max attempts	Exception propagates to channel’s error handling	SCS DLQ / error channel routing
Checkpoint after retry exhausted	Depends on channel type and checkpoint mode	NACK to AEH if max-attempts exceeded; checkpoint not committed

Test: `RequestHandlerRetryAdvice` UUID Regeneration

@SpringBootTest
@Testcontainers
class RetryAdviceUUIDTest {

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");

    @MockBean
    private StripeClient stripeClient;

    @Autowired
    private BillingServiceActivator billingServiceActivator;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    @Test
    void retryAdvice_freshUUIDPerAttempt_stripeConnectionReset_doublesCharge() {
        // Arrange: first call → StripeConnectException (Stripe committed ch_A server-side)
        //          second call (retry) → success → ch_B committed
        List<String> capturedKeys = new ArrayList<>();

        when(stripeClient.charges().create(any(), any()))
            .thenAnswer(invocation -> {
                RequestOptions opts = invocation.getArgument(1);
                capturedKeys.add(opts.getIdempotencyKey());
                if (capturedKeys.size() == 1) {
                    throw new StripeConnectException(
                        "Connection reset", null, null, 0, null);
                }
                Charge charge = mock(Charge.class);
                when(charge.getId()).thenReturn("ch_B");
                return charge;
            });

        // Act
        Message<EventData> message = buildBillingMessage(
            "acc-3", "2026-10", 14900L, 149L);
        billingServiceActivator.processBilling(message);

        // Assert: two different idempotency keys used → double charge risk
        assertThat(capturedKeys).hasSize(2);
        assertThat(capturedKeys.get(0)).isNotEqualTo(capturedKeys.get(1));
        // ch_A was committed on attempt 1 server-side but we used ch_B in our record
        assertThat(billingRecordRepo.findAll())
            .extracting(BillingRecord::getChargeId)
            .containsOnly("ch_B");
        // ch_A is orphaned on Stripe — billing mismatch
    }

    @Test
    void retryAdvice_stableContentHashKey_connectionReset_idempotent() {
        // Fixed version: both attempts use same key; Stripe returns ch_A on retry
        when(stripeClient.charges().create(any(), any()))
            .thenAnswer(invocation -> {
                RequestOptions opts = invocation.getArgument(1);
                // Attempt 1: connection reset (ch_A committed server-side)
                // Attempt 2: same key → Stripe returns idempotent response
                Charge charge = mock(Charge.class);
                when(charge.getId()).thenReturn("ch_A");
                return charge;
            });

        // Stripe will see same key on retry → idempotent, returns same ch_A

        Message<EventData> message = buildBillingMessage(
            "acc-4", "2026-10", 14900L, 149L);

        // No exception on second attempt — Stripe idempotent response
        billingServiceActivator.processBillingFixed(message);

        List<BillingRecord> records = billingRecordRepo
            .findByAccountIdAndBillingPeriod("acc-4", "2026-10");
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getChargeId()).isEqualTo("ch_A");
    }
}

Failure Mode 3: `EventHubsInboundChannelAdapter` Restart Creates a New `EventProcessorClient` That Claims the Partition Before the Old Client’s 3DS Call Completes

The Setup

The billing service uses Spring Integration’s EventHubsInboundChannelAdapter deployed in a Kubernetes pod. The adapter is configured with CheckpointMode.RECORD and a DirectChannel (both correct choices, learned from the previous two failure modes). Stripe 3DS (3D Secure) authentication is enabled for a subset of customers — international cards where the card issuer requires additional authentication. The 3DS authentication HTTP call to Stripe’s API can take 25–40 seconds for some card issuers.

@Bean
public EventHubsMessageListenerContainer listenerContainer(
        EventHubsProcessorFactory processorFactory) {
    EventHubsContainerProperties props = new EventHubsContainerProperties();
    props.setEventHubName("billing-events");
    props.setConsumerGroup("billing-consumer");
    // Correct: RECORD checkpoint fires after DirectChannel.send() completes
    props.setCheckpointConfig(new CheckpointConfig(CheckpointMode.RECORD));
    // Ownership renewal: default 30 seconds
    // (EventHubsContainerProperties.setOwnershipExpirationInterval is not set)
    return new EventHubsMessageListenerContainer(processorFactory, props);
}

@Bean
public EventHubsInboundChannelAdapter billingAdapter(
        EventHubsMessageListenerContainer container) {
    EventHubsInboundChannelAdapter adapter =
        new EventHubsInboundChannelAdapter(container);
    adapter.setOutputChannel(new DirectChannel()); // Correct: synchronous
    return adapter;
}

@Component
public class BillingServiceActivator {

    @ServiceActivator(inputChannel = "billingChannel")
    public void processBilling(Message<EventData> message) throws Exception {
        BillingEvent event = deserializeEvent(message.getPayload());

        String idempotencyKey = Hashing.sha256()
            .hashString(event.getAccountId() + "|" + event.getBillingPeriod()
                + "|" + event.getAmountCents() + "|" + event.getUnitsConsumed(),
                StandardCharsets.UTF_8).toString();

        if (billingRecordRepo.existsByAccountIdAndBillingPeriod(
                event.getAccountId(), event.getBillingPeriod())) {
            return;
        }

        // For 3DS customers, this call can take 25-40 seconds
        Charge charge = stripeClient.charges().create(
            ChargeCreateParams.builder()
                .setAmount(event.getAmountCents())
                .setCurrency("usd")
                .setCustomer(event.getStripeCustomerId())
                .build(),
            RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .setReadTimeout(60000) // 60-second timeout for 3DS
                .build()
        );

        billingRecordRepo.save(new BillingRecord(
            event.getAccountId(),
            event.getBillingPeriod(),
            charge.getId(),
            idempotencyKey
        ));
    }
}

The Kubernetes Liveness Probe and Adapter Lifecycle

The Kubernetes deployment configures a liveness probe that calls a Spring Boot Actuator health endpoint:

livenessProbe:
  httpGet:
    path: /actuator/health/liveness
    port: 8080
  initialDelaySeconds: 60
  periodSeconds: 10
  timeoutSeconds: 5
  failureThreshold: 3

The EventHubsInboundChannelAdapter’s underlying EventProcessorClient has a background thread that calls BlobClient.getLease().renew() to maintain the ownership claim on its partitions. This renewal call goes to Azure Blob Storage. If Blob Storage experiences a transient outage lasting more than 15 seconds, the renewal thread is blocked on an HTTP call for the full timeout duration.

During a Blob Storage transient outage at t=0:

t=0: Blob Storage returns HTTP 503 intermittently. The EventProcessorClient’s ownership renewal thread is stuck in an HTTP read wait.
t=0 simultaneously: Event E3 (partition 7, offset 2,201) arrives. The ServiceActivator begins a 3DS Stripe call for a customer with a German Mastercard. Stripe contacts the card issuer. Expected 3DS round-trip: 32 seconds.
t=10: The Blob Storage renewal thread is still blocked (HTTP read timeout not yet fired). The partition ownership record’s lastModifiedTime in Blob Storage has not been updated since t=-30 (the previous renewal). The ownership record is now 40 seconds old.
t=15: The Kubernetes liveness probe’s /actuator/health/liveness endpoint times out (Spring’s Actuator health check includes the EventHubsInboundChannelAdapter health, which calls the Blob Storage client to check storage availability — also blocked). Third probe failure. Kubernetes marks the pod as unhealthy.
t=15: Kubernetes sends SIGTERM to the pod. Spring’s ApplicationContext shutdown hook fires. EventHubsInboundChannelAdapter.stop() is called. The adapter calls EventHubsMessageListenerContainer.stop(), which calls EventProcessorClient.stop(). EventProcessorClient.stop() initiates a graceful shutdown — it signals the processing loop to stop and calls onPartitionClose() for each owned partition.
t=15 simultaneously: The ServiceActivator’s thread is still inside stripeClient.charges().create(), waiting for the 3DS response (E3’s 3DS call, 32-second expected duration, 17 seconds remaining).
t=15: EventProcessorClient.stop() does NOT wait for the ServiceActivator thread to complete. The MessageProducerSupport.stop() lifecycle shuts down the listening side (stops accepting new events from the partition pump), but does not interrupt the downstream processing thread. The processing thread continues running inside DirectChannel.send().
t=15: Kubernetes schedules a new pod. The new pod starts. Spring initializes a new EventHubsInboundChannelAdapter, which creates a new EventProcessorClient. The new client’s startup sequence calls BlobCheckpointStore.claimOwnership() for the partitions it wants to own. For partition 7, the ownership record’s lastModifiedTime is 40 seconds old (from before the Blob Storage outage). The ownership expiration interval is 30 seconds (default). 40 seconds > 30 seconds → ownership is expired. The new client successfully claims partition 7.
t=15: The new client’s PartitionPumpManager starts reading partition 7 from the last committed checkpoint. The last committed checkpoint for partition 7 is offset 2,200 (the offset before E3, because E3’s RECORD checkpoint never fired — E3’s ServiceActivator has not completed, so the DirectChannel.send() call has not returned, so the checkpoint has not fired on the old pod).
t=16: New pod delivers E3 (offset 2,201) to its ServiceActivator. The stable content-hash key generates the same idempotencyKey as the old pod’s attempt. BUT: the guard check — billingRecordRepo.existsByAccountIdAndBillingPeriod() — returns false because the old pod has not yet written the BillingRecord (it is still waiting on 3DS). The new pod proceeds past the guard.
t=16: New pod calls stripe.charges().create(idempotencyKey). Since the idempotency key is the same as the old pod’s in-flight call, Stripe’s idempotency layer detects the concurrent request with the same key. Stripe returns HTTP 409 (idempotency key in use, another request with the same key is currently being processed). The Stripe Java SDK throws IdempotencyException.
t=32: Old pod’s 3DS call completes. stripe.charges().create() returns ch_A. billingRecordRepo.save() writes the record. DirectChannel.send() returns. The RECORD checkpoint fires — but the old pod’s EventProcessorClient has already been stopped, so eventContext.updateCheckpoint() throws IllegalStateException("EventProcessorClient has been stopped"). The checkpoint is not committed on the old pod.
t=32: New pod’s IdempotencyException is retried by RequestHandlerRetryAdvice. stripe.charges().create(idempotencyKey) is called again. The old pod’s 3DS call has now completed, and the idempotency lock is released. Stripe returns the idempotent response for this key: ch_A (the charge committed by the old pod). The new pod saves a BillingRecord referencing ch_A. The new pod’s RECORD checkpoint fires (new EventProcessorClient is running). Partition 7 offset 2,201 is committed.

This scenario, when it works correctly (new pod gets IdempotencyException then retries with same key and receives ch_A), produces one charge. But the scenario breaks when the stable key is not used:

The Failure Sequence (Without Stable Key)

Substitute UUID.randomUUID().toString() for the content-hash key in the service activator above. The failure becomes:

t=0: Old pod: UUID.randomUUID() → UUID_A. stripe.charges().create(UUID_A) in-flight (3DS, 32s).
t=15: New pod claims partition 7 from offset 2,200. Redelivers E3.
t=16: New pod: UUID.randomUUID() → UUID_B. Guard check: existsByAccountIdAndBillingPeriod returns false (old pod hasn’t written the record yet). stripe.charges().create(UUID_B) called. No idempotency key collision (different key from UUID_A). Stripe processes as a new charge. ch_B committed at t=16+0.5s.
t=32: Old pod’s 3DS completes. stripe.charges().create(UUID_A) returns. ch_A committed. billingRecordRepo.save() writes record for ch_A. RECORD checkpoint fails (EventProcessorClient stopped).

Result: ch_A (old pod) + ch_B (new pod), both on the customer’s Stripe account. The billing database has one BillingRecord for ch_B (from the new pod) and one for ch_A (from the old pod’s save). The customer is double-charged.

Why the Guard Alone Is Not Sufficient

The guard check (existsByAccountIdAndBillingPeriod) does not protect against this failure mode because both pods check the database before either pod has written the record. The race window is:

Old pod: Stripe call started at t=0, BillingRecord NOT yet written (waiting on 3DS)
New pod: Guard check at t=16, BillingRecord not present → proceeds to charge

The guard protects against retry scenarios where the first attempt succeeded and the record was written. It does not protect against concurrent same-message processing where neither write has completed yet.

The Fix: Stable Key + Extended Ownership Expiration + Graceful Shutdown Wait

The stable content-hash key is the primary fix: both pods use the same idempotency key, so Stripe’s idempotency layer prevents two distinct charges. One of the pods will get an IdempotencyException (the concurrent-request collision), wait, and retry — receiving the idempotent ch_A response.

// Fix 1: Stable content-hash key (as shown above)
// Both pods generate the same key → Stripe prevents double charge via idempotency lock

// Fix 2: Extend ownership expiration interval to exceed maximum 3DS duration
@Bean
public EventHubsMessageListenerContainer listenerContainer(
        EventHubsProcessorFactory processorFactory) {
    EventHubsContainerProperties props = new EventHubsContainerProperties();
    props.setEventHubName("billing-events");
    props.setConsumerGroup("billing-consumer");
    props.setCheckpointConfig(new CheckpointConfig(CheckpointMode.RECORD));
    // Extend ownership expiration to 3× the maximum expected 3DS duration (40s)
    // Default is 30s — shorter than 3DS p99 latency
    props.setOwnershipExpirationInterval(Duration.ofSeconds(120));
    return new EventHubsMessageListenerContainer(processorFactory, props);
}

# Fix 3: application.yml — same property via Spring Cloud Azure configuration
spring:
  cloud:
    azure:
      eventhubs:
        processor:
          load-balancing:
            strategy: balanced
            update-interval: 10s
            ownership-expiration-interval: 120s  # >= 3× 3DS p99 latency

With ownershipExpirationInterval=120s, the new pod’s claimOwnership() call at t=15 will not succeed for partition 7 unless the ownership record is more than 120 seconds old. Since the old pod’s ownership was renewed at t=-30 (30 seconds before the Blob Storage outage), the record is 45 seconds old at t=15 — less than 120 seconds. Ownership claim fails. The new pod waits for the next load-balancing interval (default 10 seconds) before trying again. By t=32, the old pod’s 3DS call has completed and the RECORD checkpoint would have fired (if the old pod were still running). But since the old pod’s EventProcessorClient was stopped at t=15, the checkpoint fails anyway.

The complete fix therefore requires both: (1) the stable content-hash key so that concurrent same-event processing is harmless via Stripe idempotency, and (2) the extended ownership expiration interval to prevent concurrent same-event processing in the first place when the stable key is temporarily insufficient (e.g., during the window between Stripe 409 and retry).

// Fix 4: Prevent Kubernetes liveness probe from triggering on EventProcessorClient
// background thread delays. Use a dedicated liveness indicator that does NOT
// include AEH connectivity, separate from the readiness probe.

@Component
public class BillingLivenessIndicator implements LivenessStateHealthIndicator {

    @Override
    public Health health() {
        // Liveness only checks JVM-internal state (not external services)
        // AEH connectivity issues should affect Readiness, not Liveness
        return Health.up().build();
    }
}

# application.yml: exclude AEH from liveness, include in readiness only
management:
  health:
    liveness-state:
      enabled: true
    readiness-state:
      enabled: true
  endpoint:
    health:
      group:
        liveness:
          include: livenessState,diskSpace
          # Do NOT include: azure-event-hubs, blobStorage
        readiness:
          include: readinessState,db,azureEventHubs

Separating the Kubernetes liveness probe from Azure storage connectivity prevents the cascade: Blob Storage throttling → ownership renewal delay → liveness probe failure → pod restart → partition ownership race → double charge.

Graceful Shutdown: Waiting for In-Flight Processing to Complete

Spring Integration’s MessageProducerSupport.stop() does not wait for downstream processing to complete by default. Adding a SmartLifecycle.stop(Runnable) callback that waits for the processing thread is a deeper fix:

@Bean
public EventHubsInboundChannelAdapter billingAdapter(
        EventHubsMessageListenerContainer container) {
    EventHubsInboundChannelAdapter adapter =
        new EventHubsInboundChannelAdapter(container) {
            @Override
            public void stop(Runnable callback) {
                // Wait up to 90 seconds for in-flight processing to complete
                // before stopping the container (releasing partition ownership)
                waitForInFlightProcessing(Duration.ofSeconds(90));
                super.stop(callback);
            }
        };
    adapter.setOutputChannel(new DirectChannel());
    return adapter;
}

Alternatively, configure the pod’s terminationGracePeriodSeconds in the Kubernetes deployment to exceed the maximum 3DS duration, and set Spring Boot’s graceful shutdown timeout accordingly:

# Kubernetes deployment spec
spec:
  template:
    spec:
      terminationGracePeriodSeconds: 120  # >= 3x 3DS p99 latency (40s)
      containers:
        - name: billing-service
          lifecycle:
            preStop:
              exec:
                command: ["/bin/sh", "-c", "sleep 5"]  # Allow SIGTERM to propagate

# application.yml
server:
  shutdown: graceful
spring:
  lifecycle:
    timeout-per-shutdown-phase: 90s  # Match terminationGracePeriodSeconds - 5

Test: Adapter Restart Race During 3DS

@SpringBootTest
@Testcontainers
class AdapterRestartMid3DSTest {

    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");

    @MockBean
    private StripeClient stripeClient;

    @Autowired
    private ApplicationContext applicationContext;

    @Autowired
    private BillingRecordRepository billingRecordRepo;

    @Test
    void adapterRestart_mid3DS_stableKey_stripeIdempotencyPreventsDoubleCharge()
            throws Exception {
        CountDownLatch oldPodStarted = new CountDownLatch(1);
        CountDownLatch newPodStarted = new CountDownLatch(1);
        CountDownLatch oldPod3DSComplete = new CountDownLatch(1);

        AtomicInteger stripeCallCount = new AtomicInteger(0);

        when(stripeClient.charges().create(any(), any()))
            .thenAnswer(invocation -> {
                int n = stripeCallCount.incrementAndGet();
                RequestOptions opts = invocation.getArgument(1);
                String key = opts.getIdempotencyKey();

                if (n == 1) {
                    // Old pod: simulate 32-second 3DS call
                    oldPodStarted.countDown();
                    newPodStarted.await(5, TimeUnit.SECONDS); // wait for new pod
                    Thread.sleep(200); // simulate remaining 3DS time
                    Charge charge = mock(Charge.class);
                    when(charge.getId()).thenReturn("ch_A");
                    oldPod3DSComplete.countDown();
                    return charge;
                } else if (n == 2) {
                    // New pod attempt 1: concurrent request → 409
                    throw new IdempotencyException(
                        "Another request with this idempotency key is in progress",
                        key, null, 409, null);
                } else {
                    // New pod attempt 2 (after retry): ch_A idempotent response
                    Charge charge = mock(Charge.class);
                    when(charge.getId()).thenReturn("ch_A");
                    return charge;
                }
            });

        BillingEvent event = buildBillingEvent("acc-5", "2026-10", 29900L, 299L);

        // Simulate old pod processing E3 with stable key
        CompletableFuture.runAsync(() -> {
            processEventOnOldPod(event);
        });

        oldPodStarted.await(2, TimeUnit.SECONDS);

        // Simulate new pod claiming partition and redelivering E3
        newPodStarted.countDown();
        processEventOnNewPod(event); // retries on 409, gets ch_A idempotent

        oldPod3DSComplete.await(5, TimeUnit.SECONDS);

        // Assert: only one BillingRecord — ch_A — no double charge
        List<BillingRecord> records = billingRecordRepo
            .findByAccountIdAndBillingPeriod("acc-5", "2026-10");
        assertThat(records).hasSize(1);
        assertThat(records.get(0).getChargeId()).isEqualTo("ch_A");
        assertThat(stripeCallCount.get()).isEqualTo(3); // old attempt, new attempt, retry
    }
}

Spring Integration vs. Other Azure Event Hubs Integration Layers

The three failure modes above arise specifically from Spring Integration’s component model. For reference, here is how each mode manifests across different Azure Event Hubs integration approaches:

Failure mode	Raw EventProcessorClient	SCS AEH Binder	Spring Integration AEH
Checkpoint before processing completes	Not possible — developer controls checkpoint explicitly	BATCH mode only (batch function completes before checkpoint)	Any async channel type: ExecutorChannel, QueueChannel, PriorityChannel
Fresh UUID per retry attempt	Only if developer codes it that way in onEvent()	SCS max-attempts retries full Consumer function — same issue	RequestHandlerRetryAdvice retries full handleRequestMessage() — same issue
Adapter restart mid-3DS	EventProcessorClient.stop() does not wait for onEvent() callback	SCS binder stop() also does not wait for Consumer function	MessageProducerSupport.stop() does not wait for downstream processing
Primary defense	Stable key + guard + manual checkpoint ordering	Stable key + guard + checkpoint mode selection	Stable key + guard + DirectChannel + ownershipExpirationInterval

The Two-Layer Defense for Spring Integration Azure Event Hubs

The two-layer defense pattern applies across all three failure modes. Layer one (at the Keybrake proxy boundary) prevents uncapped Stripe spending; layer two (in-service idempotency) prevents duplicate charges from reaching Stripe at all.

Layer One: Keybrake Scoped Key with Spend Cap

Replace the live Stripe key in the service activator with a Keybrake-scoped key that is locked to charges:write on a single customer in a single billing period. Configure a per-billing-period spend cap equal to the maximum expected charge amount. The Keybrake proxy rejects any Stripe API call that would exceed the cap — so if both the old pod and the new pod send valid Stripe calls for the same customer, only one succeeds at the proxy layer.

// application.yml
keybrake:
  stripe:
    base-url: https://api.keybrake.com/stripe
    key-template: "keybrake_per_period_${account_id}_${billing_period}"
    spend-cap-cents: 100000  # max charge per account per billing period

@ServiceActivator
public void processBilling(Message<EventData> message) {
    BillingEvent event = deserializeEvent(message.getPayload());

    String idempotencyKey = Hashing.sha256()
        .hashString(event.getAccountId() + "|" + event.getBillingPeriod()
            + "|" + event.getAmountCents() + "|" + event.getUnitsConsumed(),
            StandardCharsets.UTF_8).toString();

    if (billingRecordRepo.existsByAccountIdAndBillingPeriod(
            event.getAccountId(), event.getBillingPeriod())) {
        return;
    }

    // Keybrake-scoped key: proxy enforces single charge per account per period
    String scopedApiKey = keybrakeClient.issueKey(
        KeyScope.builder()
            .stripeOperation("charges:write")
            .customer(event.getStripeCustomerId())
            .spendCapCents(event.getAmountCents())
            .ttlSeconds(3600)
            .build()
    );

    Charge charge = Stripe.charges().create(
        buildParams(event),
        RequestOptions.builder()
            .setApiKey(scopedApiKey)  // Keybrake proxy key
            .setIdempotencyKey(idempotencyKey)
            .build()
    );

    billingRecordRepo.save(new BillingRecord(
        event.getAccountId(), event.getBillingPeriod(),
        charge.getId(), idempotencyKey
    ));
}

Layer Two: In-Service Defense (Stable Key + Guard + DirectChannel)

Layer two requires three elements working together:

Stable content-hash idempotency key — same value on every retry, redelivery, or concurrent processing attempt for the same billing event. Never use UUID.randomUUID(), message offset, sequence number, or any header that varies between deliveries of the same event.
BillingRecord guard check — query the database for an existing record before calling Stripe. This defends against the case where the previous attempt succeeded (committed ch_A, wrote the record) but the exception was thrown in a post-Stripe step. The guard skips the Stripe call entirely on retry when the record already exists.
DirectChannel as the adapter’s output channel — ensures the RECORD checkpoint fires only after the entire downstream processing chain (including the Stripe call and the guard write) completes without exception. Any async channel type breaks this guarantee.

Diagnostic Checklist for Spring Integration Azure Event Hubs + Stripe

What type is the adapter’s output channel? If not DirectChannel, checkpoint mode RECORD will fire before processing completes. Either switch to DirectChannel or switch checkpoint mode to MANUAL.
Where is the idempotency key generated? If inside the @ServiceActivator method body using UUID.randomUUID(), it is regenerated on every RequestHandlerRetryAdvice attempt. Move key generation to a stable content hash or a HeaderEnricher upstream of the retry scope.
What is ownershipExpirationInterval? Default is 30 seconds. If Stripe 3DS p99 latency exceeds this, partition ownership can be claimed by a new pod while the old pod’s Stripe call is in flight. Set to 2 × Stripe 3DS p99 latency.
Does the Kubernetes liveness probe include Azure storage connectivity? If yes, Blob Storage throttling will trigger pod restarts, leading to the adapter restart failure mode. Separate liveness (JVM-internal state) from readiness (external service connectivity).
What is the pod’s terminationGracePeriodSeconds? If less than the maximum Stripe API call duration (including 3DS), in-flight Stripe calls will be cut off mid-request on pod shutdown. Set to maximum Stripe call duration + 30s buffer.
Is RequestHandlerRetryAdvice configured with retryOn(StripeConnectException.class)? If yes, verify the idempotency key is stable (content hash) so Stripe’s idempotency layer handles concurrent calls from old and new pods correctly.
Is there a BillingRecord guard check before the Stripe call? The guard must be in the same transactional boundary as the guard write after the Stripe call — not just a pre-Stripe query in isolation. Use @Transactional(isolation = SERIALIZABLE) on the guard + write if concurrent processing from multiple pods is possible.
Is CheckpointMode.MANUAL used with ExecutorChannel? Verify that checkpointer.updateCheckpoint(message).block() is called after (not before) the billingRecordRepo.save() call. The Checkpointer header value from the EventHubsInboundChannelAdapter is available in the AzureHeaders.CHECKPOINTER message header and can be called from any thread that has access to the Message object.

Summary

Spring Integration’s EventHubsInboundChannelAdapter exposes the same underlying Azure Event Hubs checkpoint and partition ownership machinery as the Spring Cloud Stream AEH Binder and the raw EventProcessorClient, but through a different abstraction layer that introduces unique failure modes. ExecutorChannel decouples checkpoint timing from processing completion — RECORD checkpoint fires when the message is queued, not when the downstream handler returns. RequestHandlerRetryAdvice re-invokes the entire service activator method body on each retry, including any UUID.randomUUID() calls that generate fresh idempotency keys. EventHubsInboundChannelAdapter restart via Kubernetes liveness probe creates a race between the old pod’s in-flight 3DS call and the new pod’s immediate partition ownership claim.

All three failure modes share a common root cause: the Stripe idempotency key is not stable across the invocations that produce duplicate charges. A content-hash key derived from the billing event’s business fields — accountId, billingPeriod, amountCents, unitsConsumed — produces the same value in every invocation context. Stripe’s idempotency layer then converts concurrent or sequential duplicate invocations into a single charge. The BillingRecord guard check adds defense in depth for the case where the stable key alone is insufficient (the two-pod concurrent scenario during the window before either pod has written the guard record). The Keybrake scoped proxy key adds a third layer at the API boundary, capping spending per account per billing period regardless of in-service state.

Scoped Stripe keys for Spring Integration billing pipelines

Keybrake issues per-billing-period Stripe API keys with spend caps — a proxy layer that rejects double charges before they reach Stripe, regardless of which Spring Integration channel type, retry advice, or adapter lifecycle behavior triggers them. Works with any JVM billing service.

Spring Integration Azure Event Hubs Architecture

How This Differs from Spring Cloud Stream AEH Binder

Failure Mode 1: ExecutorChannel Commits the RECORD Checkpoint at Queue Time, Not at Processing Completion Time

The Setup

The Failure Sequence

The Developer Misconception

The Fix: Use DirectChannel for Billing Flows, or Switch to MANUAL Checkpoint

Test: Verifying ExecutorChannel Checkpoint Timing

Failure Mode 2: RequestHandlerRetryAdvice Re-invokes the Entire Service Activator Including UUID.randomUUID() on Each Retry Attempt

The Setup

How RequestHandlerRetryAdvice Works

The Failure Sequence

The Developer Misconception

The Fix: Stable Content-Hash Key + BillingRecord Guard

Alternative: Generate the Key Upstream in a HeaderEnricher

Comparison: Spring Integration Advice Chain vs. SCS AEH Binder Retry

Test: RequestHandlerRetryAdvice UUID Regeneration

Failure Mode 3: EventHubsInboundChannelAdapter Restart Creates a New EventProcessorClient That Claims the Partition Before the Old Client’s 3DS Call Completes

The Setup

The Kubernetes Liveness Probe and Adapter Lifecycle

The Failure Sequence (Without Stable Key)

Why the Guard Alone Is Not Sufficient

The Fix: Stable Key + Extended Ownership Expiration + Graceful Shutdown Wait

Graceful Shutdown: Waiting for In-Flight Processing to Complete

Test: Adapter Restart Race During 3DS

Spring Integration vs. Other Azure Event Hubs Integration Layers

The Two-Layer Defense for Spring Integration Azure Event Hubs

Layer One: Keybrake Scoped Key with Spend Cap

Layer Two: In-Service Defense (Stable Key + Guard + DirectChannel)

Diagnostic Checklist for Spring Integration Azure Event Hubs + Stripe

Summary

Scoped Stripe keys for Spring Integration billing pipelines

Failure Mode 1: `ExecutorChannel` Commits the RECORD Checkpoint at Queue Time, Not at Processing Completion Time

The Fix: Use `DirectChannel` for Billing Flows, or Switch to `MANUAL` Checkpoint

Test: Verifying `ExecutorChannel` Checkpoint Timing

Failure Mode 2: `RequestHandlerRetryAdvice` Re-invokes the Entire Service Activator Including `UUID.randomUUID()` on Each Retry Attempt

How `RequestHandlerRetryAdvice` Works

The Fix: Stable Content-Hash Key + `BillingRecord` Guard

Alternative: Generate the Key Upstream in a `HeaderEnricher`

Test: `RequestHandlerRetryAdvice` UUID Regeneration

Failure Mode 3: `EventHubsInboundChannelAdapter` Restart Creates a New `EventProcessorClient` That Claims the Partition Before the Old Client’s 3DS Call Completes