Spring Integration and Spring Batch Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Spring Batch’s FaultTolerantStep retry mechanism wraps ItemWriter.write() invocations in a RetryTemplate. When write() throws a retryable exception, RetryTemplate re-invokes the callback — calling write() again on the same chunk. UUID.randomUUID() inside write() is a call expression that evaluates when write() executes, so it fires fresh on every retry attempt. The initial write creates ch_A before a StripeException wrapping a socket timeout; the first retry invocation evaluates a new UUID.randomUUID(), causing Stripe to create ch_B. Three Spring Integration and Spring Batch-specific Stripe billing failure modes: FaultTolerantStep retry re-invokes ItemWriter.write() with fresh UUID.randomUUID() per attempt — subtler: FaultTolerantChunkProcessor’s scatter-gather item scan re-calls write() up to N times per chunk failure before retry even fires, each scan call generating a distinct UUID; Spring Integration’s RequestHandlerRetryAdvice re-invokes handleRequestMessage() with fresh UUID.randomUUID() per retry — subtler: using MessageHeaders.ID as the idempotency key is safe on same-message retry but unsafe when a MessagePublishingErrorHandler republishes the payload as a new Message with a new ID UUID; and TaskExecutorPartitionHandler concurrent partition TOCTOU race plus job-restart re-execution of failed chunks — both paths evaluate UUID.randomUUID() freshly against customers whose ch_A was already committed at Stripe.

This post covers all three failure modes with Java code (Spring Batch 5.x and Spring Integration 6.x), content-hash idempotency keys stable across write() retry invocations and handleRequestMessage() retry invocations, pg_try_advisory_lock() for cross-pod job serialization, and pre-flight PostgreSQL ON CONFLICT DO NOTHING as a cluster-wide billing mutex — plus per-billing-period vault keys via a spend-cap proxy as a hard financial backstop. For Spring Boot application-level @Retryable annotations, see the Spring Boot and Stripe Integration post. For reactive Spring WebFlux retry on Mono and Flux, see the Spring WebFlux and Stripe Integration post.

Failure mode 1: FaultTolerantStep retry re-invokes ItemWriter.write() — UUID.randomUUID() inside write() evaluates fresh per invocation — initial write creates ch_A before StripeException — first retry creates ch_B

Spring Batch’s fault-tolerant step processes items in chunks. A chunk is read from the ItemReader, optionally transformed by the ItemProcessor, and written in a single ItemWriter.write(Chunk<? extends O> chunk) call. When you configure FaultTolerantStepBuilder.retry(StripeException.class).retryLimit(3), Spring Batch wraps the write() invocation in a RetryTemplate. When write() throws a StripeException that matches the retry policy, RetryTemplate.execute(RetryCallback) calls the callback again — which calls write() again.

The callback does not change between attempts. It is the same functional reference. But a method call is not a memoized result: write() is called again, and any call expression inside write(), including UUID.randomUUID(), evaluates on every invocation:

// UNSAFE: UUID.randomUUID() inside ItemWriter.write() body.
// Spring Batch FaultTolerantStep retry re-invokes write() on every retry attempt.
// The initial invocation and every subsequent retry each generate a distinct UUID.

import com.stripe.model.Charge;
import com.stripe.net.RequestOptions;
import com.stripe.param.ChargeCreateParams;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;

import java.util.UUID;

@Component
@StepScope
public class StripeChargingItemWriter implements ItemWriter<BillingItem> {

    @Override
    public void write(Chunk<? extends BillingItem> chunk) throws Exception {
        for (BillingItem item : chunk) {
            // UNSAFE: evaluated per write() invocation, not per chunk assembly.
            // Attempt 1: UUID = "4c8a3b1d-2e5f-4b7e-9c0a-1a2b3c4d5e6f" → ch_A
            // Attempt 2: UUID = "d7e2f4a6-3b8c-4f1e-0d9a-2b3c4d5e6f7a" → ch_B ← duplicate
            String idempotencyKey = UUID.randomUUID().toString();

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(idempotencyKey)
                .build();
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(item.getAmountCents())
                .setCurrency("usd")
                .setCustomer(item.getCustomerId())
                .build();
            Charge.create(params, options);
        }
    }
}

// Step configuration:
@Bean
public Step billingStep(JobRepository jobRepository,
                        PlatformTransactionManager txManager,
                        StripeChargingItemWriter writer) {
    return new StepBuilder("billingStep", jobRepository)
        .<BillingItem, BillingItem>chunk(50, txManager)
        .reader(billingItemReader())
        .writer(writer)
        .faultTolerant()
        .retry(com.stripe.exception.StripeException.class)
        .retryLimit(3)
        .build();
}

// Execution:
// Chunk of 50 customers read and passed to write().
// Customer 23: POST /v1/charges → StripeException (socket timeout, 30s).
//   Stripe committed ch_A before timeout fired.
// Spring Batch RetryTemplate catches StripeException, waits backoff, calls write() again.
// Customer 23: idempotencyKey = NEW UUID → POST /v1/charges → ch_B committed.
// write() returns success. Job records 50 items written.
// Two charges exist for customer 23 for this billing period.

The @StepScope annotation on StripeChargingItemWriter does not protect against this failure. @StepScope creates one bean instance per StepExecution — it scopes the bean’s lifecycle, not the number of times write() is called within that execution. Spring Batch retry calls write() multiple times on the same bean instance. An engineer who reads @StepScope and infers “one UUID per step run” is misreading what the annotation does.

Subtler variant: FaultTolerantChunkProcessor scatter-gather scan re-calls write() up to N times per chunk failure before retry even fires

Spring Batch’s FaultTolerantChunkProcessor implements a more sophisticated failure isolation mechanism than a simple retry loop. When write() throws a retryable exception on a chunk of N items, the chunk processor does not immediately retry the full chunk. Instead it enters a “scan” phase: it calls write() with each item individually (chunk size = 1) to determine which item caused the failure, so that the bad item can be skipped if a SkipPolicy is configured.

This scan phase calls write() up to N times before the retry policy even fires on the original chunk-level exception. If write() calls UUID.randomUUID() per item per invocation, and the chunk has 50 customers, a single StripeException on customer 23 causes write() to be called up to 49 more times during the scan (once per remaining item to check). Each of those scan calls generates a distinct UUID.randomUUID() per customer, even for customers whose charges succeeded in the original chunk write. Customer 7, which was successfully charged as ch_A in the original bulk write(), gets a scan invocation of write() with a new UUID — Stripe creates ch_B for customer 7 even though the original charge succeeded:

// What FaultTolerantChunkProcessor actually does when write(chunk[50]) throws:
//
// 1. write(chunk[50]) → StripeException on item 23 (ch_A for items 1-22 already committed)
// 2. Scan phase: individual write() calls to isolate the bad item:
//    write([item_1])  → UUID_1b → Stripe: ch_B for customer 1  ← duplicate!
//    write([item_2])  → UUID_2b → Stripe: ch_B for customer 2  ← duplicate!
//    ...
//    write([item_22]) → UUID_22b → Stripe: ch_B for customer 22 ← duplicate!
//    write([item_23]) → StripeException again → item_23 is the bad item
//    write([item_24]) → UUID_24a → no prior charge, ok
//    ...
//    write([item_50]) → UUID_50a → no prior charge, ok
// 3. RetryPolicy: write() threw on item_23 again → retry count 1
// 4. Retry: write(filtered_chunk[49]) → still uses UUID per item call → more fresh UUIDs
//
// Result: customers 1-22 each have TWO charges for this billing period.
// This happens before the retry counter even increments.

// The fix is identical: stable key derived outside write(), not inside:

@Component
@StepScope
public class StripeChargingItemWriter implements ItemWriter<BillingItem> {

    @Override
    public void write(Chunk<? extends BillingItem> chunk) throws Exception {
        for (BillingItem item : chunk) {
            // SAFE: key derived from deterministic inputs that do not change between
            // write() invocations, scan phases, or retry attempts.
            // sha256(customerId:billingPeriod:spring-batch-billing) produces the same
            // 32-char hex string for the same customer in the same billing period
            // regardless of which write() invocation or scan iteration this is.
            String stableKey = DigestUtils.sha256Hex(
                item.getCustomerId() + ":" +
                item.getBillingPeriod() + ":" +
                "spring-batch-billing"
            ).substring(0, 32);

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(stableKey)
                .build();
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(item.getAmountCents())
                .setCurrency("usd")
                .setCustomer(item.getCustomerId())
                .build();
            Charge.create(params, options);
        }
    }
}

The fix applies identically to ItemProcessor.process() if the idempotency key is generated there. process() is re-invoked on each retry attempt in the same way that write() is. The stable key must be derived from deterministic inputs that are available in the item itself (customer ID, billing period) — not from UUID.randomUUID(), System.currentTimeMillis(), or any per-invocation state.

Using a job-parameter-derived component in the key is safe when those parameters are set once at job launch and do not change between restart attempts. stepExecution.getJobParameters().getString("billingPeriod") returns the same value across all write invocations and all restart attempts for a given JobInstance. Combining it with the customer ID produces a key that is unique per customer per billing period and stable across all retry and scan invocations:

// Alternative stable key using StepExecution job parameters (safe on job restart too).
// billingPeriod job parameter set once at JobLauncher.run() call time.

@Component
@StepScope
public class StripeChargingItemWriter implements ItemWriter<BillingItem> {

    @Value("#{stepExecution}")
    private StepExecution stepExecution;

    @Override
    public void write(Chunk<? extends BillingItem> chunk) throws Exception {
        // Derive stable suffix from job parameters — deterministic across all
        // write() invocations, scan phases, retries, and job restarts.
        String billingPeriod = stepExecution.getJobParameters()
            .getString("billingPeriod"); // e.g. "2026-10"

        for (BillingItem item : chunk) {
            String stableKey = DigestUtils.sha256Hex(
                item.getCustomerId() + ":" + billingPeriod + ":spring-batch-billing"
            ).substring(0, 32);

            RequestOptions options = RequestOptions.builder()
                .setIdempotencyKey(stableKey)
                .build();
            ChargeCreateParams params = ChargeCreateParams.builder()
                .setAmount(item.getAmountCents())
                .setCurrency("usd")
                .setCustomer(item.getCustomerId())
                .build();
            Charge.create(params, options);
        }
    }
}

There is one important check: Stripe’s idempotency cache uses the key to match requests only within a 24-hour window. A job that starts in October 2026, fails mid-chunk, and is restarted 25 hours later will get a cache miss for the stable key on the same customer — Stripe will process the charge again. For billing periods longer than 24 hours (monthly billing), the stable key approach still prevents the within-window duplicate from the scatter-gather scan and immediate retry, but does not prevent the cross-window duplicate from a delayed restart. The pre-flight ON CONFLICT DO NOTHING database guard (covered in failure mode 3) closes that gap by checking a durable billing ledger regardless of Stripe’s cache TTL.

Failure mode 2: RequestHandlerRetryAdvice re-invokes handleRequestMessage() — UUID.randomUUID() inside the handler fires per retry — initial handling creates ch_A — first retry creates ch_B

Spring Integration’s RequestHandlerRetryAdvice decorates a MessageHandler with a RetryTemplate. When the handler throws a retryable exception, the advice re-invokes the handler’s handleRequestMessage(message) method via the retry callback. The message argument is the same Message object on every retry invocation — Spring Integration does not create a new message for the retry. But UUID.randomUUID() inside handleRequestMessage() is not reading from the message; it is a call expression in the method body that evaluates when the method executes:

// UNSAFE: UUID.randomUUID() inside AbstractReplyProducingMessageHandler.handleRequestMessage().
// Spring Integration RequestHandlerRetryAdvice re-invokes handleRequestMessage()
// on every retry attempt triggered by the RetryTemplate.
// The initial invocation and every subsequent retry each generate a distinct UUID.

import com.stripe.model.Charge;
import com.stripe.net.RequestOptions;
import com.stripe.param.ChargeCreateParams;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.messaging.Message;

import java.util.UUID;

@Component
public class StripeBillingHandler extends AbstractReplyProducingMessageHandler {

    @Override
    protected Object handleRequestMessage(Message<?> requestMessage) {
        BillingPayload payload = (BillingPayload) requestMessage.getPayload();

        // UNSAFE: evaluated per handleRequestMessage() invocation.
        // RequestHandlerRetryAdvice re-invokes this method on each retry.
        // Attempt 1: UUID = "a1b2c3d4-..." → POST /v1/charges → StripeException
        //            (network timeout; ch_A committed at Stripe before exception)
        // Attempt 2: UUID = "e5f6a7b8-..." → POST /v1/charges → ch_B ← duplicate
        String idempotencyKey = UUID.randomUUID().toString();

        RequestOptions options = RequestOptions.builder()
            .setIdempotencyKey(idempotencyKey)
            .build();
        ChargeCreateParams params = ChargeCreateParams.builder()
            .setAmount(payload.getAmountCents())
            .setCurrency("usd")
            .setCustomer(payload.getCustomerId())
            .build();

        try {
            Charge charge = Charge.create(params, options);
            return new BillingResult(charge.getId(), "succeeded");
        } catch (Exception e) {
            throw new RuntimeException("Stripe charge failed", e);
        }
    }
}

// Spring Integration flow configuration:
@Bean
public IntegrationFlow billingFlow() {
    return IntegrationFlow
        .from("billingInputChannel")
        .handle(stripeBillingHandler(), h -> h.advice(retryAdvice()))
        .channel("billingOutputChannel")
        .get();
}

@Bean
public RequestHandlerRetryAdvice retryAdvice() {
    RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
    RetryTemplate retryTemplate = RetryTemplate.builder()
        .maxAttempts(3)
        .fixedBackoff(500)
        .retryOn(RuntimeException.class)
        .build();
    advice.setRetryTemplate(retryTemplate);
    return advice;
}

// Execution:
// Message arrives at billingInputChannel: {customerId="cust_abc", amount=4999}
// handleRequestMessage() invoked: idempotencyKey="a1b2c3d4-..."
//   → POST /v1/charges → StripeException (socket timeout, 30s)
//   → ch_A committed at Stripe before timeout
// RequestHandlerRetryAdvice catches RuntimeException, waits 500ms
// handleRequestMessage() re-invoked with same Message: idempotencyKey="e5f6a7b8-..."
//   → POST /v1/charges → ch_B committed ← duplicate charge

Subtler variant: using MessageHeaders.ID as idempotency key — safe on same-message retry, but a MessagePublishingErrorHandler republishes the payload as a new Message with a new ID UUID

Some engineers address the UUID-in-handler problem by switching the idempotency key source from UUID.randomUUID() to message.getHeaders().getId().toString(). Spring Integration assigns a UUID to every Message at creation time as MessageHeaders.ID. When RequestHandlerRetryAdvice retries the handler, it passes the same Message object to each re-invocation — so headers.getId() returns the same UUID on all retry attempts within that advice’s retry loop. This is safe for the direct retry path.

The failure occurs on the error recovery path. When the retry advice exhausts all attempts, Spring Integration’s error handling can route the failure to an error channel. A common pattern is to use MessagePublishingErrorHandler with a DelayedRetryPublisher or a simple republish-to-input-channel strategy. When the error handler republishes the original payload to billingInputChannel for a delayed retry, Spring Integration creates a new Message wrapping the same payload — but with a new MessageHeaders.ID UUID assigned at creation time:

// Pattern that appears safe but breaks on error-channel republish:

@Component
public class StripeBillingHandler extends AbstractReplyProducingMessageHandler {

    @Override
    protected Object handleRequestMessage(Message<?> requestMessage) {
        BillingPayload payload = (BillingPayload) requestMessage.getPayload();

        // Appears safe: MessageHeaders.ID is stable across RequestHandlerRetryAdvice
        // retry invocations because the same Message object is replayed.
        // But MessageHeaders.ID changes when the message is republished to the channel
        // by error handling after retry exhaustion.
        String idempotencyKey = requestMessage.getHeaders().getId().toString();

        // ...
    }
}

// Error handler that republishes after delay:
@Bean
public MessageChannel billingErrorChannel() {
    return MessageChannels.direct().getObject();
}

@Bean
public IntegrationFlow errorHandlingFlow() {
    return IntegrationFlow
        .from("billingErrorChannel")
        .transform(errorMessage -> {
            // Extract original payload from ErrorMessage
            Message<?> failedMsg = ((MessagingException)
                ((ErrorMessage) errorMessage).getPayload()).getFailedMessage();
            return failedMsg.getPayload(); // BillingPayload extracted
        })
        .delay(d -> d.defaultDelay(60_000L)) // wait 60s before re-attempting
        .channel("billingInputChannel") // republish to input channel
        .get();
}

// Execution:
// Message M1 (headers.id = UUID-1): arrives at billingInputChannel
// handleRequestMessage(M1) → attempt 1: key = "UUID-1" → ch_A committed, StripeException
// handleRequestMessage(M1) → attempt 2: key = "UUID-1" → Stripe returns 200 (cached)
// handleRequestMessage(M1) → attempt 3: key = "UUID-1" → Stripe returns 200 (cached)
// All 3 retries used UUID-1 → Stripe's idempotency cache returns ch_A each time. SAFE.
//
// BUT: if errorHandlingFlow republishes the payload:
// errorHandlingFlow creates new Message M2 (headers.id = UUID-2) from BillingPayload
// handleRequestMessage(M2) → key = "UUID-2" → ch_B committed ← duplicate!
// The error path bypasses Stripe's idempotency cache because the key changed.

The fix is to store the stable idempotency key in a user-defined header at message origination time, so it is preserved even when the payload is republished as a new Message:

// SAFE: stable key stored in a named MessageHeader at message creation time.
// The key survives republish because the error handler explicitly preserves it.

// At message origination (before sending to billingInputChannel):
String stableKey = DigestUtils.sha256Hex(
    payload.getCustomerId() + ":" +
    payload.getBillingPeriod() + ":" +
    "si-billing"
).substring(0, 32);

Message<BillingPayload> message = MessageBuilder
    .withPayload(payload)
    .setHeader("X-Billing-Idempotency-Key", stableKey)
    .build();
messagingTemplate.send("billingInputChannel", message);

// Handler reads from header — safe across all retry invocations and republishes:
@Component
public class StripeBillingHandler extends AbstractReplyProducingMessageHandler {

    @Override
    protected Object handleRequestMessage(Message<?> requestMessage) {
        BillingPayload payload = (BillingPayload) requestMessage.getPayload();

        // SAFE: reads from user-defined header set at message creation.
        // Same value on every RequestHandlerRetryAdvice retry invocation
        // AND on every error-channel republish that preserves this header.
        String idempotencyKey = (String) requestMessage.getHeaders()
            .get("X-Billing-Idempotency-Key");

        if (idempotencyKey == null) {
            // Defensive: fall back to deterministic computation if header missing.
            idempotencyKey = DigestUtils.sha256Hex(
                payload.getCustomerId() + ":" +
                payload.getBillingPeriod() + ":" +
                "si-billing"
            ).substring(0, 32);
        }

        RequestOptions options = RequestOptions.builder()
            .setIdempotencyKey(idempotencyKey)
            .build();
        // ...
    }
}

// Error handler: explicitly copy the stable key header when republishing:
@Bean
public IntegrationFlow errorHandlingFlow() {
    return IntegrationFlow
        .from("billingErrorChannel")
        .transform(errorMessage -> {
            Message<?> failedMsg = ((MessagingException)
                ((ErrorMessage) errorMessage).getPayload()).getFailedMessage();
            // Preserve stable key header on republish:
            return MessageBuilder
                .withPayload(failedMsg.getPayload())
                .copyHeadersIfAbsent(failedMsg.getHeaders())
                .build();
        })
        .delay(d -> d.defaultDelay(60_000L))
        .channel("billingInputChannel")
        .get();
}

Subtler variant: MessageHandlerChain with retry advice — a transformer upstream in the chain that computes UUID.randomUUID() fires per retry chain re-run

When RequestHandlerRetryAdvice is applied to a MessageHandlerChain, the advice wraps the entire chain, not just the terminal handler. When the terminal handler throws and the retry advice fires, the entire chain re-runs — including any MessageTransformingHandler earlier in the chain. A transformer that adds an Idempotency-Key header by calling UUID.randomUUID() fires again on every retry chain re-run, overwriting the header with a fresh UUID before handleRequestMessage() reads it:

// UNSAFE: transformer inside the chain adds Idempotency-Key header via UUID.randomUUID().
// RequestHandlerRetryAdvice wraps the chain — re-runs the entire chain on retry.
// The transformer fires again → overwrites header with fresh UUID → ch_B on retry.

@Bean
public IntegrationFlow billingChainFlow() {
    return IntegrationFlow
        .from("billingInputChannel")
        .handle(MessageHandlerChain.builder()
            .handler(idempotencyKeyTransformer()) // UNSAFE: adds header via UUID.randomUUID()
            .handler(stripeBillingHandler())      // reads header set by transformer
            .get(),
            h -> h.advice(retryAdvice()))        // advice wraps the chain
        .channel("billingOutputChannel")
        .get();
}

// The transformer is the problem:
@Bean
public MessageTransformingHandler idempotencyKeyTransformer() {
    return new MessageTransformingHandler(message -> {
        // UNSAFE: fires on initial chain run AND on every retry chain re-run.
        return MessageBuilder.fromMessage(message)
            .setHeader("Idempotency-Key", UUID.randomUUID().toString()) // fresh per run
            .build();
    });
}

// Fix: compute stable key and store it BEFORE the message enters the chain —
// in the message origination code, not inside the chain transformer.
// The transformer becomes a no-op guard (reads existing header, never overwrites):
@Bean
public MessageTransformingHandler idempotencyKeyTransformer() {
    return new MessageTransformingHandler(message -> {
        if (message.getHeaders().containsKey("X-Billing-Idempotency-Key")) {
            return message; // already set — do not overwrite on retry re-run
        }
        // Only set if missing (first run, not a retry chain re-run).
        // Prefer computing stable key from payload over UUID.randomUUID():
        BillingPayload payload = (BillingPayload) message.getPayload();
        String stableKey = DigestUtils.sha256Hex(
            payload.getCustomerId() + ":" + payload.getBillingPeriod() + ":si-billing"
        ).substring(0, 32);
        return MessageBuilder.fromMessage(message)
            .setHeader("X-Billing-Idempotency-Key", stableKey)
            .build();
    });
}

The conditional guard works because RequestHandlerRetryAdvice replays the same Message object on each retry chain re-run. The header set by the first transformer run is present on the replayed Message. The conditional check prevents the transformer from overwriting it with a fresh UUID. This pattern also handles the case where the chain is not wrapped with retry advice but a caller-level retry loop creates a new Message per attempt: the fallback stable-key computation inside the transformer produces the same value from the same payload inputs.

Failure mode 3: TaskExecutorPartitionHandler TOCTOU race — concurrent partition threads compute distinct UUID.randomUUID() per customer — ch_A from partition 1, ch_B from partition 2 — plus job-restart re-execution of failed chunks with fresh UUIDs

Spring Batch’s partition step dispatches multiple StepExecution instances to a TaskExecutor thread pool via TaskExecutorPartitionHandler. Each partition runs a full step lifecycle — read, process, write — concurrently on a thread pool thread. If two partition slices contain the same customer due to a partitioning logic error (non-disjoint ranges, off-by-one in range boundaries, or a query that selects based on status before any partition marks billing as started), both partitions’ write() calls attempt to charge the same customer concurrently:

// Partitioner that produces non-disjoint customer ranges under concurrent execution.
// Two partitions both read customer cust_xyz if the partitioner queries billing_status
// before either partition updates it.

@Component
public class CustomerBillingPartitioner implements Partitioner {

    private final JdbcTemplate jdbc;

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        // UNSAFE: reads all pending customers at partition time.
        // If two pods both call JobLauncher.run() for the same billing period
        // before either pod updates billing_status, both get overlapping customer lists.
        List<String> pendingCustomers = jdbc.queryForList(
            "SELECT customer_id FROM customers WHERE billing_status = 'pending' " +
            "AND billing_period = ?",
            String.class,
            billingPeriod
        );

        Map<String, ExecutionContext> partitions = new LinkedHashMap<>();
        int partitionSize = (int) Math.ceil((double) pendingCustomers.size() / gridSize);
        for (int i = 0; i < gridSize; i++) {
            int from = i * partitionSize;
            int to = Math.min(from + partitionSize, pendingCustomers.size());
            if (from >= pendingCustomers.size()) break;
            ExecutionContext ctx = new ExecutionContext();
            ctx.put("customerIds", pendingCustomers.subList(from, to));
            ctx.put("partitionId", "partition-" + i);
            partitions.put("partition-" + i, ctx);
        }
        return partitions;
    }
}

// Step configuration:
@Bean
public Step partitionedBillingStep(JobRepository jobRepository,
                                   PlatformTransactionManager txManager) {
    return new StepBuilder("partitionedBillingStep", jobRepository)
        .partitioner("billingWorkerStep", customerBillingPartitioner())
        .partitionHandler(taskExecutorPartitionHandler())
        .build();
}

@Bean
public TaskExecutorPartitionHandler taskExecutorPartitionHandler() {
    TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
    handler.setTaskExecutor(new SimpleAsyncTaskExecutor());
    handler.setStep(billingWorkerStep());
    handler.setGridSize(4); // 4 concurrent partition threads
    return handler;
}

// Execution with unsafe ItemWriter (UUID inside write()):
// Partition 0: write([cust_abc, ...]) → idempotencyKey for cust_abc = UUID_thread0
// Partition 1: write([cust_abc, ...]) → idempotencyKey for cust_abc = UUID_thread1
// (cust_abc appears in both partitions due to TOCTOU on billing_status query)
// Both reach Stripe: ch_A from thread 0, ch_B from thread 1 ← duplicate charge

The stable-key fix prevents the duplicate charge even when both partitions contain the same customer. If write() derives the key from sha256(customerId:billingPeriod:spring-batch-billing)[:32], both partition threads produce the same key for cust_abc. Stripe’s idempotency cache serializes concurrent requests with the same key — one returns a 200 with the charge, the other returns the same 200 (the cached response). No duplicate charge is created.

But the stable key is not a substitute for fixing the TOCTOU race in the partitioner. The TOCTOU race means both partitions are doing redundant work: both attempt to charge cust_abc, one wins the idempotency cache race, the other blocks on the Stripe API waiting for the first to complete. Under load with 500 customers, this doubles the Stripe API call volume and the wall-clock time of the billing run. The correct fix combines the stable key (financial safety) with a pre-flight database guard (operational efficiency):

// SAFE ItemWriter: stable key + pre-flight ON CONFLICT DO NOTHING guard.
// The stable key prevents duplicate charges even if two partitions include the same customer.
// The ON CONFLICT guard prevents redundant Stripe API calls for already-billed customers.

@Component
@StepScope
public class SafeStripeChargingItemWriter implements ItemWriter<BillingItem> {

    private final JdbcTemplate jdbc;
    private final String billingPeriod;

    public SafeStripeChargingItemWriter(JdbcTemplate jdbc,
            @Value("#{stepExecution.jobParameters['billingPeriod']}") String billingPeriod) {
        this.jdbc = jdbc;
        this.billingPeriod = billingPeriod;
    }

    @Override
    public void write(Chunk<? extends BillingItem> chunk) throws Exception {
        for (BillingItem item : chunk) {
            String stableKey = DigestUtils.sha256Hex(
                item.getCustomerId() + ":" + billingPeriod + ":spring-batch-billing"
            ).substring(0, 32);

            // Pre-flight: atomically claim this billing slot.
            // INSERT ... ON CONFLICT DO NOTHING returns 0 if another partition
            // or pod already started billing this customer this period.
            int inserted = jdbc.update(
                "INSERT INTO billing_ledger (customer_id, billing_period, idempotency_key, status) " +
                "VALUES (?, ?, ?, 'in_progress') " +
                "ON CONFLICT (customer_id, billing_period) DO NOTHING",
                item.getCustomerId(), billingPeriod, stableKey
            );

            if (inserted == 0) {
                // Another partition already claimed this customer. Skip.
                continue;
            }

            try {
                RequestOptions options = RequestOptions.builder()
                    .setIdempotencyKey(stableKey)
                    .build();
                ChargeCreateParams params = ChargeCreateParams.builder()
                    .setAmount(item.getAmountCents())
                    .setCurrency("usd")
                    .setCustomer(item.getCustomerId())
                    .build();
                Charge charge = Charge.create(params, options);

                jdbc.update(
                    "UPDATE billing_ledger SET status = 'completed', " +
                    "charge_id = ?, completed_at = NOW() " +
                    "WHERE customer_id = ? AND billing_period = ?",
                    charge.getId(), item.getCustomerId(), billingPeriod
                );
            } catch (StripeException e) {
                // Mark failed — the stable key means a retry will hit Stripe's
                // idempotency cache and return ch_A if it was committed, preventing ch_B.
                jdbc.update(
                    "UPDATE billing_ledger SET status = 'failed' " +
                    "WHERE customer_id = ? AND billing_period = ?",
                    item.getCustomerId(), billingPeriod
                );
                throw new RuntimeException("Stripe charge failed for " + item.getCustomerId(), e);
            }
        }
    }
}

Job-restart re-execution of failed chunks: FaultTolerantStepBuilder re-runs uncommitted items with fresh UUID.randomUUID() per restart

Spring Batch’s job restart mechanism is a second, distinct failure path that the stable key must cover. When a StepExecution is marked FAILED (due to exceeding skipLimit, a non-skippable exception, or a JVM crash), calling JobLauncher.run(job, params) with the same JobParameters resumes from the failed step. Spring Batch’s JobRepository stores the read-item offset in the ExecutionContext — the restarted run picks up reading from where the failed run stopped and re-processes the uncommitted chunk.

If the original step execution committed ch_A for customers 1–180 before failing at customer 181, the restarted run re-reads customers from the reader offset and passes them to write() again. For customers 181 onward, no charge exists — the restart correctly charges them. But if the batch job used a restart-unsafe ItemReader that re-reads customers starting from the beginning rather than from the stored offset, customers 1–180 are re-written by the restart write() call with fresh UUID.randomUUID() per customer — ch_B for each:

// Restart failure scenario: restart-unsafe ItemReader rewinds to beginning.
// This happens when:
// 1. JdbcCursorItemReader used without saveState=true (default is true, but
//    if ExecutionContext is not persisted due to crash before flush, reader rewinds).
// 2. Custom ItemReader that does not implement ItemStream and save read position.
// 3. JdbcPagingItemReader without a stable sort key — page boundaries shift on restart.

// Customers 1-180: previously charged (ch_A through ch_CKP) — stable key required.
// On restart, if reader re-reads from page 1:
// write([cust_1, cust_2, ...]) with UUID per customer → ch_B per customer ← duplicates

// The pre-flight ON CONFLICT DO NOTHING guard prevents the restart duplicate:
// INSERT for cust_1 returns 0 (billing_ledger row exists with status=completed) → skip.
// All already-completed customers are skipped in O(1) per INSERT.

// For customers 181+ (not yet billed):
// INSERT returns 1 (no existing row) → proceed to Stripe → stable key → ch_A.
// Stable key ensures even a partial chunk failure within the restarted step
// does not create ch_B via FaultTolerantChunkProcessor scan.

Cross-pod job duplication: two pods both launch the same JobInstance before the JobRepository uniqueness constraint fires

Spring Batch’s JobRepository enforces JobInstance uniqueness via a database unique constraint on (job_name, job_key), where job_key is a hash of the JobParameters. If two pods call JobLauncher.run(job, sameParams) simultaneously — before either pod’s INSERT INTO batch_job_instance commits — both pods may pass the uniqueness check at the application layer and both attempt to insert. The database constraint fires for the second insertion, causing a JobInstanceAlreadyCompleteException or JobExecutionAlreadyRunningException on the second pod.

In practice, the uniqueness constraint usually protects against this. The dangerous scenario is when billing jobs are triggered by consuming a message from a queue (SQS, RabbitMQ, Kafka) and the message is delivered to multiple consumers simultaneously or delivered twice. The second delivery triggers a second JobLauncher.run()` call, which either hits the uniqueness constraint (if the first job completed) or creates a new `JobExecution` for the same `JobInstance` (running state). The duplicate `JobExecution` runs the step again, including `write()` calls, generating fresh UUIDs per customer. The pg_try_advisory_lock() cross-pod mutex serializes job launches at the application layer before the JobRepository insertion:

// SAFE: pg_try_advisory_lock() serializes concurrent job launches across pods.
// The advisory lock uses a hash of the billing period string as the lock key,
// so two pods launching the same billing job will contend on the same lock.

@Service
public class BillingJobLauncherService {

    private final JobLauncher jobLauncher;
    private final Job billingJob;
    private final JdbcTemplate jdbc;

    public void launchBillingJob(String billingPeriod) throws Exception {
        long lockKey = Math.abs(("spring-batch-monthly-billing:" + billingPeriod).hashCode());

        // Try to acquire the advisory lock. Returns false immediately if another
        // session holds it — no blocking, no deadlock risk.
        Boolean acquired = jdbc.queryForObject(
            "SELECT pg_try_advisory_lock(?)",
            Boolean.class,
            lockKey
        );

        if (!acquired) {
            log.info("Billing job for {} already running on another pod — skipping launch",
                billingPeriod);
            return;
        }

        try {
            JobParameters params = new JobParametersBuilder()
                .addString("billingPeriod", billingPeriod)
                .toJobParameters();
            jobLauncher.run(billingJob, params);
        } finally {
            // Release lock after job completes (or fails).
            // Advisory locks are session-scoped — also released on connection close.
            jdbc.execute("SELECT pg_advisory_unlock(" + lockKey + ")");
        }
    }
}

The lock scope is the database session. If the pod crashes mid-job, PostgreSQL releases the advisory lock when the connection closes, allowing another pod to acquire it and launch the job. The stable idempotency key and the ON CONFLICT DO NOTHING pre-flight guard together ensure the restarted job does not create duplicate charges for customers already billed in the crashed execution.

Summary table

Implementation checklist for Spring Batch + Spring Integration + Stripe billing

• Never call UUID.randomUUID() inside ItemWriter.write(), ItemProcessor.process(), or any Spring Integration MessageHandler.handleRequestMessage(). Both FaultTolerantStep retry and RequestHandlerRetryAdvice re-invoke these methods on each retry attempt. The key must be derived from deterministic inputs available in the item payload (customer ID, billing period) before the method body executes.
• Account for FaultTolerantChunkProcessor scatter-gather scan. When write() fails on a chunk, Spring Batch may call write() individually for each item in the chunk to identify the failing item. A safe stable key handles scan invocations the same way it handles retry invocations — the same key for the same customer on every write() call regardless of context.
• Do not use MessageHeaders.ID as the Stripe idempotency key in Spring Integration flows that have error-channel republish logic. MessageHeaders.ID is stable within a RequestHandlerRetryAdvice retry loop (same Message object) but changes when the payload is republished as a new Message. Use a user-defined header set from a content-hash of payload fields, preserved by copyHeadersIfAbsent() on republish.
• Apply the stable key fix to MessageHandlerChain transformers, not just the terminal handler. If a transformer upstream in the chain adds an idempotency header via UUID.randomUUID(), the chain’s retry advice will re-run the transformer and overwrite the header with a fresh UUID before the terminal handler reads it. The transformer must either compute a stable content-hash key or guard against overwriting an existing header on retry chain re-runs.
• Add a pre-flight INSERT ... ON CONFLICT (customer_id, billing_period) DO NOTHING as the authoritative cluster-wide billing mutex. The stable key prevents duplicate Stripe charges at the idempotency cache layer, but the pre-flight database guard prevents redundant Stripe API calls in the first place — important when TaskExecutorPartitionHandler creates overlapping partitions or when a job is restarted with a reader that rewinds to the beginning.
• Set pg_try_advisory_lock() before JobLauncher.run() to serialize cross-pod job launches. Spring Batch’s JobRepository uniqueness constraint fires at the database layer, but the time window between two pods passing the application-layer check and hitting the constraint is enough for both to create separate JobExecution instances and start running the step concurrently. The advisory lock serializes the launch at the application layer before any JobRepository state is written.
• Add a vault key spend cap at expected_total × 1.10. The cap fires at the proxy layer regardless of what Spring Batch retry configuration, Spring Integration error handling, or partition handler configuration is deployed. It is the last line of defense after all code-level idempotency fixes, and it bounds the financial damage if any combination of the above failure modes fires simultaneously during a billing run.