BatchBuilder

Builders for processing collections of items as coordinated batch jobs. Ratchet provides two batch builders:

• BatchBuilder -- for in-memory collections where the total size is known up front.
• StreamingBatchBuilder<T> -- for large datasets read from a Stream, processed in chunks without loading everything into memory.

Package: run.ratchet.api

BatchBuilder

Obtained from JobSchedulerService.enqueueBatch().

forEach

<T extends Serializable> BatchBuilder forEach(
    Collection<T> items, SerializableConsumer<T> action)

Applies an action to each item in the collection. Each item becomes a child job in the batch.

Type Parameters:

• T -- the type of elements; must implement Serializable.

Parameters:

• items -- the collection of items to process.
• action -- the operation to perform on each item.

Returns: the BatchBuilder for chaining.

List<Long> orderIds = List.of(1L, 2L, 3L, 4L, 5L);

scheduler.enqueueBatch("Process Orders")
    .forEach(orderIds, orderId -> orderService.process(orderId))
    .withMaxRetries(3)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(2))
    .submit();

withMaxRetries

BatchBuilder withMaxRetries(int retries)

Sets the number of retry attempts for every child job. The default is 0, so a child runs once unless retries are explicitly enabled. A negative value throws IllegalArgumentException.

The setting belongs to the whole builder and applies even when called after forEach. It does not apply to the no-op batch parent or workflow branches.

withBackoff

BatchBuilder withBackoff(BackoffPolicy policy, Duration param)

Sets the backoff policy and base delay used between child retry attempts. Both arguments are required. This does not enable retries by itself; set withMaxRetries to a positive value as well. The setting applies to the whole builder, regardless of call order.

scheduler.enqueueBatch("Process Orders")
    .forEach(orderIds, orderId -> orderService.process(orderId))
    .withMaxRetries(3)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(2))
    .submit();

withTimeout

BatchBuilder withTimeout(Duration timeout)

Sets the execution timeout for every child job. Null is the default and clears the override, leaving children on the global default SLA. The setting applies to the whole builder, including children already added. It does not apply to the batch parent or to workflow branch and callback jobs.

Timeouts are stored in whole seconds. Negative durations, positive durations under one second, and values above Integer.MAX_VALUE whole seconds throw IllegalArgumentException.

scheduler.enqueueBatch("Process Orders")
    .forEach(orderIds, orderId -> orderService.process(orderId))
    .withTimeout(Duration.ofMinutes(45))
    .submit();

withCallbackOptions

BatchBuilder withCallbackOptions(JobOptions options)

Sets priority, retries, backoff and timeout for jobs created by thenOnBatchSuccess, thenOnBatchFailure, thenWhenSuccessRate, thenWhenFailureCount, thenWhenBatch and thenBranch.

The default is JobOptions.defaults(): NORMAL priority, no retries, no backoff and the global default SLA. Child settings from withMaxRetries, withBackoff and withTimeout do not apply to callbacks. The setting applies to the whole builder regardless of call order. Callbacks use these options for both retries and backoff.

batch.withCallbackOptions(JobOptions.defaults()
    .withMaxRetries(2)
    .withTimeout(Duration.ofHours(1))
    .withPriority(JobPriority.HIGH));

withResource

BatchBuilder withResource(String resourceName)

Every child acquires a permit from the named resource pool before it runs. At capacity, children are rescheduled without counting as retries. The configured permit count caps how many children run at once. That count is shared with all other jobs using the same resource name, inside or outside the batch.

The setting applies to every child regardless of call order. The no-op parent, progress hooks, and workflow branch jobs do not inherit it. Null or blank disables resource limiting. Other names are trimmed.

submit() throws UnsupportedOperationException if a resource is set and the store does not advertise the ResourcePermitStore capability. Streaming batches have the same behavior at start().

Configure permit counts and retry delays in scheduler_resource_limit, using max_concurrent and retry_delay_ms. See Performance Tuning.

scheduler.enqueueBatch("Payments")
    .forEach(paymentIds, id -> paymentService.charge(id))
    .withResource("payment-api")
    .submit();

onProgress

BatchBuilder onProgress(SerializableConsumer<BatchContext> hook)

Registers a progress hook invoked as child jobs complete. The hook receives a BatchContext snapshot with current progress metrics.

Parameters:

• hook -- a consumer receiving BatchContext updates.

Returns: the BatchBuilder for chaining.

scheduler.enqueueBatch("Import Records")
    .forEach(records, record -> importService.importRecord(record))
    .onProgress(ctx -> {
        log.info("Batch {}: {}% complete ({}/{} items, {} failed)",
            ctx.batchId(), ctx.percentDone(),
            ctx.completedItems(), ctx.totalItems(), ctx.failedItems());
    })
    .submit();

thenOnBatchSuccess

BatchBuilder thenOnBatchSuccess(SerializableCheckedRunnable next)

Schedules a job to execute when all child jobs complete successfully (zero failures).

Parameters:

• next -- the task to execute on batch success.

scheduler.enqueueBatch("Nightly Sync")
    .forEach(accounts, acct -> syncService.sync(acct))
    .thenOnBatchSuccess(() -> notificationService.sendSyncComplete())
    .submit();

thenOnBatchFailure

BatchBuilder thenOnBatchFailure(SerializableCheckedRunnable next)

Schedules a job to execute when one or more child jobs fail.

Parameters:

• next -- the task to execute on batch failure.

scheduler.enqueueBatch("Data Migration")
    .forEach(rows, row -> migrationService.migrate(row))
    .thenOnBatchFailure(() -> alertService.migrationPartiallyFailed())
    .submit();

thenWhenBatch

BatchBuilder thenWhenBatch(
    SerializablePredicate<BatchContext> condition,
    SerializableCheckedRunnable next)

Schedules a job when a custom condition on the BatchContext is met.

Parameters:

• condition -- predicate evaluating the BatchContext.
• next -- the task to execute when the condition is true.

public final class BatchWorkflowConditions {
    public static boolean isCompleteWithoutFailures(BatchContext ctx) {
        return ctx.failedItems() == 0 && ctx.isComplete();
    }

    public static boolean shouldRollback(BatchContext ctx) {
        return ctx.failedItems() > ctx.totalItems() / 2;
    }
}

scheduler.enqueueBatch("Process Items")
    .forEach(items, item -> processItem(item))
    .thenWhenBatch(BatchWorkflowConditions::isCompleteWithoutFailures,
                   () -> archiveResults())
    .thenWhenBatch(BatchWorkflowConditions::shouldRollback,
                   () -> rollbackProcessing())
    .submit();

Custom batch predicates are analyzed into JobPayload JSON at submission time. Put compound logic in a public helper or CDI bean method, then pass a method reference.

thenWhenSuccessRate

BatchBuilder thenWhenSuccessRate(double minRate, SerializableCheckedRunnable next)

Schedules a job when the success rate meets or exceeds the specified threshold.

Parameters:

• minRate -- minimum success rate (0.0 to 1.0).
• next -- the task to execute when the success rate condition is met.

scheduler.enqueueBatch("Email Campaign")
    .forEach(recipients, r -> emailService.sendCampaign(r))
    .thenWhenSuccessRate(0.95, () -> log.info("Campaign delivered successfully"))
    .thenWhenSuccessRate(0.50, () -> alertService.campaignPartialFailure())
    .submit();

thenWhenFailureCount

BatchBuilder thenWhenFailureCount(int maxFailures, SerializableCheckedRunnable next)

Schedules a job when the number of failures stays at or below the specified threshold (failures less than or equal to maxFailures).

Parameters:

• maxFailures -- the maximum number of failures (inclusive) for which the action still fires.
• next -- the task to execute when the failure count is within the threshold.

scheduler.enqueueBatch("Import Data")
    .forEach(rows, row -> importRow(row))
    .thenWhenFailureCount(10, () -> alertService.importWithinFailureBudget())
    .submit();

thenBranch

BatchBuilder thenBranch(WorkflowCondition condition,
                        SerializableCheckedRunnable next,
                        String description)

Adds a workflow branch with an explicit WorkflowCondition and description.

Parameters:

• condition -- the WorkflowCondition determining when this branch fires.
• next -- the task to execute.
• description -- human-readable description for monitoring.

public final class BatchWorkflowConditions {
    public static boolean hasMoreThanFiveFailures(BatchContext ctx) {
        return ctx.failedItems() > 5 && ctx.isComplete();
    }
}

scheduler.enqueueBatch("Complex Batch")
    .forEach(items, item -> processItem(item))
    .thenBranch(
        WorkflowCondition.batchCustom(BatchWorkflowConditions::hasMoreThanFiveFailures),
        () -> escalateToOps(),
        "Escalate when more than 5 items fail")
    .submit();

submit

JobHandle submit()

Submits the configured batch for execution.

Returns: a JobHandle for the batch parent job.

JobHandle handle = scheduler.enqueueBatch("My Batch")
    .forEach(items, item -> processItem(item))
    .submit();

log.info("Batch submitted with ID {}", handle.id());

StreamingBatchBuilder

Obtained from JobSchedulerService.streamingBatch(). Designed for large datasets where items are read from a Stream and inserted in configurable chunks.

Framework extensions that build invocations directly (rather than serializing lambdas) have mirrored builders — InvocationBatchBuilder and InvocationStreamingBatchBuilder<T> — on the InvocationSubmissionService SPI.

StreamingBatchBuilder exposes the same child retry, timeout, resource, and callback option methods as BatchBuilder:

StreamingBatchBuilder<T> withMaxRetries(int retries)
StreamingBatchBuilder<T> withBackoff(BackoffPolicy policy, Duration param)
StreamingBatchBuilder<T> withTimeout(Duration timeout)
StreamingBatchBuilder<T> withCallbackOptions(JobOptions options)
StreamingBatchBuilder<T> withResource(String resourceName)

The invocation-typed mirrors expose the same methods and persist the same child retry options and resource name.

fromStream

StreamingBatchBuilder<T> fromStream(Stream<T> stream)

Sets the input data source for the batch. The stream item type T is fixed when the builder is created via streamingBatch() and must implement Serializable.

Parameters:

• stream -- the stream of items to process.

scheduler.<Long>streamingBatch("Process Users")
    .fromStream(userRepository.streamAllIds())
    // ...

process

StreamingBatchBuilder<T> process(SerializableCheckedConsumer<T> action)

Configures the processing logic applied to each item. The action can throw checked exceptions.

Parameters:

• action -- the processing action for each item.

scheduler.<Long>streamingBatch("Migrate Users")
    .fromStream(userIds.stream())
    .process(userId -> migrationService.migrateUser(userId))
    // ...

withChunkSize

StreamingBatchBuilder<T> withChunkSize(int size)

Sets the number of items per database insert chunk. Default is 100. Tune this based on your database's bulk insert performance.

Parameters:

• size -- items per chunk. Must be positive.

scheduler.<Record>streamingBatch("Bulk Import")
    .fromStream(records.stream())
    .process(record -> importRecord(record))
    .withChunkSize(2000)
    // ...

onProgress

StreamingBatchBuilder<T> onProgress(Consumer<StreamingBatchContext> hook)

Registers a progress hook called during the stream consumption phase (not during job execution). Receives a StreamingBatchContext with streaming progress.

Parameters:

• hook -- a consumer receiving StreamingBatchContext updates.

scheduler.<Long>streamingBatch("Stream Import")
    .fromStream(dataStream)
    .process(item -> processItem(item))
    .onProgress(ctx -> log.info("Streamed {} items in {} chunks",
        ctx.processedItems(), ctx.chunksInserted()))
    // ...

onBatchProgress

StreamingBatchBuilder<T> onBatchProgress(SerializableConsumer<BatchContext> hook)

Registers a progress hook called during job execution (after streaming is complete). Receives a BatchContext with execution progress.

Parameters:

• hook -- a consumer receiving BatchContext updates.

scheduler.<Long>streamingBatch("Process Stream")
    .fromStream(ids.stream())
    .process(id -> processId(id))
    .onBatchProgress(ctx -> log.info("Execution: {}% complete", ctx.percentDone()))
    // ...

Workflow methods

StreamingBatchBuilder supports the same workflow methods as BatchBuilder:

StreamingBatchBuilder<T> thenOnBatchSuccess(SerializableCheckedRunnable next)
StreamingBatchBuilder<T> thenOnBatchFailure(SerializableCheckedRunnable next)
StreamingBatchBuilder<T> thenWhenBatch(
    SerializablePredicate<BatchContext> condition, SerializableCheckedRunnable next)
StreamingBatchBuilder<T> thenWhenFailureCount(int maxFailures, SerializableCheckedRunnable next)
StreamingBatchBuilder<T> thenWhenSuccessRate(double minRate, SerializableCheckedRunnable next)

start

JobHandle start()

Starts the streaming batch operation. The stream is consumed, items are inserted in chunks, and child jobs are created.

Returns: a JobHandle for the batch parent job.

JobHandle handle = scheduler.<Long>streamingBatch("Full Migration")
    .fromStream(repository.streamAll())
    .process(id -> migrationService.migrate(id))
    .withChunkSize(1000)
    .withMaxRetries(3)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(2))
    .onProgress(ctx -> log.info("Streamed {} items", ctx.processedItems()))
    .thenOnBatchSuccess(() -> log.info("Migration complete"))
    .thenOnBatchFailure(() -> alertService.migrationFailed())
    .start();

BatchBuilder vs StreamingBatchBuilder

Aspect	BatchBuilder	StreamingBatchBuilder
Input	Collection<T> (in memory)	Stream<T> (lazy)
Memory	Entire collection loaded	Chunked, constant memory
Total known at start	Yes	No (stream not exhausted)
Progress during creation	N/A	StreamingBatchContext
Progress during execution	BatchContext	BatchContext
Best for	Small-medium collections	Large datasets, database cursors
Submit method	submit()	start()

See also

• BatchContext Reference
• WorkflowCondition Reference
• JobSchedulerService Reference