JobBuilder

Fluent builder API for creating and configuring individual jobs. JobBuilder is obtained from JobSchedulerService.enqueue() or schedule() and provides methods for retry policies, priorities, timeouts, workflows, callbacks, tags, and parameters.

Package: run.ratchet.apiType: Interface

Basic usage

JobHandle handle = scheduler.enqueue(() -> orderService.process(orderId))
    .withPriority(JobPriority.HIGH)
    .withMaxRetries(3)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(10))
    .withTimeout(Duration.ofMinutes(5))
    .withTags("order-processing", "customer-123")
    .withParam("orderId", String.valueOf(orderId))
    .onSuccess(ctx -> log.info("Order {} processed", orderId))
    .onFailure((ctx, error) -> alertService.sendAlert(error))
    .submit();

Configuration methods

withPriority

JobBuilder withPriority(JobPriority priority)

Sets the execution priority. Higher priority jobs are executed before lower priority ones when multiple jobs are queued.

Parameters:

• priority -- the priority level. One of LOWEST, LOW, NORMAL (default), HIGH, CRITICAL.

scheduler.enqueue(() -> criticalService.process())
    .withPriority(JobPriority.CRITICAL)
    .submit();

withMaxRetries

JobBuilder withMaxRetries(int retries)

Sets the maximum number of retry attempts after failure. The default is 0 (no retries).

Parameters:

• retries -- maximum retry count. Must be non-negative.

scheduler.enqueue(() -> externalApi.call())
    .withMaxRetries(5)
    .submit();

withBackoff

JobBuilder withBackoff(BackoffPolicy policy, Duration param)

Configures the backoff strategy applied between retry attempts. Only relevant when maxRetries > 0.

Parameters:

• policy -- the backoff strategy: NONE (immediate retry), FIXED (constant delay), or EXPONENTIAL (doubling delay).
• param -- the base delay duration. For FIXED, this is the constant delay. For EXPONENTIAL, this is the initial delay.

// Exponential backoff starting at 2 seconds
scheduler.enqueue(() -> paymentService.charge(amount))
    .withMaxRetries(4)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(2))
    .submit();
// Retries at: 2s, 4s, 8s, 16s

withTimeout

JobBuilder withTimeout(Duration timeout)

Sets the maximum execution duration. Jobs exceeding this limit are forcibly terminated and marked as failed.

Parameters:

• timeout -- the maximum execution duration. Use Duration.ZERO to disable timeout.

scheduler.enqueue(() -> longRunningImport.execute())
    .withTimeout(Duration.ofMinutes(30))
    .submit();

withTags

JobBuilder withTags(String... tags)

Adds one or more tags to the job. Tags are trimmed, converted to lowercase, and stored only if non-null and non-blank. Tags enable filtering and categorization.

Parameters:

• tags -- one or more tag strings.

scheduler.enqueue(() -> reportService.generate(reportId))
    .withTags("reports", "finance", "q4-2024")
    .submit();

withParam

JobBuilder withParam(String key, String value)

Adds a key-value parameter accessible during execution via JobContext.param(). Parameters provide lightweight configuration without serializing complex objects.

Parameters:

• key -- parameter key. Must not be null or blank.
• value -- parameter value. Must not be null.

scheduler.enqueue(() -> emailService.send())
    .withParam("recipient", "[email protected]")
    .withParam("template", "welcome")
    .submit();

// Inside the job:
// String recipient = JobContext.current().param("recipient");

withResource

JobBuilder withResource(String resourceName)

Specifies a shared resource that this job requires. The job will acquire a permit from the resource pool before execution. If no permits are available, the job is rescheduled with a delay.

This limits concurrent access to shared resources regardless of job type (e.g., capping concurrent API calls to a payment gateway at 5 total).

Parameters:

• resourceName -- the name of the resource to acquire. If null or blank, no resource limiting is applied.

scheduler.enqueue(() -> paymentGateway.charge(paymentId))
    .withResource("payment-api")
    .submit();

withIdempotencyKey

JobBuilder withIdempotencyKey(String key)

Overrides the auto-generated idempotency key. By default, a UUID is generated at builder creation time. A custom idempotency key is globally unique -- once used, that key is consumed forever, preventing duplicate job creation.

Parameters:

• key -- the idempotency key. If null or blank, keeps the auto-generated UUID.

Difference from withBusinessKey:

• idempotencyKey is UNIQUE globally -- once used, that key is consumed forever.
• businessKey only blocks active (PENDING/RUNNING) jobs -- allows re-runs after completion.

// Webhook handler: same delivery ID = same job forever
scheduler.enqueue(() -> webhookHandler.process(payload))
    .withIdempotencyKey(webhookDeliveryId)
    .submit();

withBusinessKey

JobBuilder withBusinessKey(String key)

Sets a business key for preventing concurrent execution against the same entity. Unlike withIdempotencyKey, the business key allows multiple completed jobs with the same key over time -- it only blocks when an active (PENDING/RUNNING) job exists with the same key.

Parameters:

• key -- the business key. If null or blank, no concurrent execution blocking is applied. After trimming, it may contain up to 255 printable ASCII characters (U+0020 through U+007E). Ratchet rejects longer or non-ASCII keys instead of leaving a store to truncate or convert them.

// Only one sync per user at a time, re-runs allowed after completion
scheduler.enqueue(() -> syncService.syncUser(userId))
    .withBusinessKey("sync-user-" + userId)
    .submit();

immediate

JobBuilder immediate()

Marks the job for immediate execution notification. The scheduler publishes a wakeup to all cluster nodes, bypassing the normal adaptive polling delay.

Jobs with CRITICAL priority or zero delay are automatically treated as immediate. Use this method explicitly when you need immediate behavior for other configurations.

scheduler.enqueue(() -> urgentService.handle(alert))
    .immediate()
    .submit();

virtual

JobBuilder virtual()

Routes the job to the virtual-thread executor pool (ExecutorTargets.VIRTUAL). Mutually exclusive with platform(); the last call wins. If you call neither, the job runs on the deployment's default threading mode.

A target selects a pool, not a guarantee about thread type: the container decides what backs each pool. If no virtual executor is configured, a virtual-targeted job falls back to the platform pool, recorded with a metric and a one-time warning. Because a node only claims work it has a pool for, execution targets also shape which jobs a node picks up.

scheduler.enqueue(() -> externalApi.fetchAll(batchId))
    .virtual()
    .submit();

platform

JobBuilder platform()

Routes the job to the platform-thread executor pool (ExecutorTargets.PLATFORM), which is always present. Mutually exclusive with virtual(); the last call wins.

scheduler.enqueue(() -> reportService.render(reportId))
    .platform()
    .submit();

awaitSignal

JobBuilder awaitSignal(String signalKey, Duration timeout)

Creates the job in WAITING status instead of making it immediately eligible for execution. The job stays blocked until a matching signal is delivered through JobSchedulerService.deliverSignal(). If the signal is not delivered before timeout, Ratchet fails the job with SignalTimeoutException.

Parameters:

• signalKey -- named signal used for broadcast delivery. Use stable domain keys such as order:123:approved.
• timeout -- maximum time to wait for the signal. Must be positive.

scheduler.enqueue(() -> approvalService.continueOrder(orderId))
    .awaitSignal("order:" + orderId + ":approved", Duration.ofHours(24))
    .withTags("approval", "orders")
    .submit();

// Later, from an admin action or another workflow:
scheduler.deliverSignal(
    "order:" + orderId + ":approved",
    SignalDecision.approved("approved-by-manager"));

Inside the waiting job, read the payload through JobContext.signalPayload():

public void continueOrder(UUID orderId) {
    SignalDecision decision = JobContext.current().signalPayload(SignalDecision.class);
    if (decision != null && decision.isRejected()) {
        throw new OrderRejectedException(decision.rejectionReason());
    }
    fulfillment.start(orderId);
}

withEncryptedPayload

JobBuilder withEncryptedPayload()

Opts this job in to payload encryption at rest. Its protected surfaces, the job payload and result, are encrypted before they are stored, while routing and bookkeeping columns stay in cleartext so the job is still claimable and queryable. This is the per-job equivalent of the deployment-wide switch on RatchetOptions, described in the Payload Encryption guide.

Encryption takes effect only when the deployment has a PayloadEncryption engine and a KeyProvider installed. Without them, an opted-in job fails when it is submitted rather than persisting data it was asked to protect. Enabling encryption globally with nothing installed is a startup-time failure instead.

scheduler.enqueue(() -> billingService.charge(cardToken))
    .withEncryptedPayload()
    .submit();

See the Payload Encryption guide for setup, engine choice, and key rotation.

Callback methods

onSuccess

JobBuilder onSuccess(SerializableConsumer<JobContext> s)

Registers a callback invoked after successful job completion. The callback receives the JobContext of the completed job.

Parameters:

• s -- success callback accepting a JobContext.

scheduler.enqueue(() -> importService.importData(batchId))
    .onSuccess(ctx -> log.info("Job {} completed", ctx.jobId()))
    .submit();

onFailure

JobBuilder onFailure(SerializableBiConsumer<JobContext, Throwable> f)

Registers a callback invoked if the job fails. The callback receives both the JobContext and the Throwable that caused the failure.

Parameters:

• f -- failure callback accepting a JobContext and Throwable.

scheduler.enqueue(() -> riskyService.execute())
    .onFailure((ctx, error) -> {
        alertService.page("Job " + ctx.jobId() + " failed: " + error.getMessage());
    })
    .submit();

Workflow methods

then

JobBuilder then(SerializableCheckedRunnable next)

Adds a task to the sequential chain. Chain tasks execute in order after the primary task completes successfully. If any task in the chain fails, subsequent tasks are not executed (unless configured with workflow branches).

Parameters:

• next -- the next task in the chain. Must not be null.

scheduler.enqueue(() -> validateData())
    .then(() -> processData())
    .then(() -> generateReport())
    .then(() -> sendNotification())
    .submit();

thenOnSuccess

JobBuilder thenOnSuccess(SerializableCheckedRunnable next)

Schedules a separate job to execute if the current job succeeds. This creates a workflow branch with a SUCCESS condition. Unlike then(), this creates an independent job, not a chain step.

Parameters:

• next -- the task to execute on success as a separate job.

scheduler.enqueue(() -> processPayment(orderId))
    .thenOnSuccess(() -> sendReceipt(orderId))
    .thenOnFailure(() -> refundPayment(orderId))
    .submit();

thenOnFailure

JobBuilder thenOnFailure(SerializableCheckedRunnable next)

Schedules a separate job to execute if the current job fails (after all retries are exhausted). This creates a workflow branch with a FAILURE condition.

Parameters:

• next -- the task to execute on failure as a separate job.

scheduler.enqueue(() -> processPayment(orderId))
    .thenOnFailure(() -> alertTeam(orderId))
    .submit();

when

<T> JobBuilder when(
    SerializablePredicate<JobResult<T>> condition,
    SerializableCheckedRunnable next)

Schedules a job to execute when a custom condition is met. The condition receives the full JobResult<T> of the current job. Workflow routing is exclusive: the first matching branch runs and all remaining sibling branches are canceled. Equal priorities preserve registration order.

Type Parameters:

• T -- the type of the job result.

Parameters:

• condition -- predicate evaluated against the JobResult.
• next -- the task to execute when the condition is true.

public final class DataAnalysisConditions {
    public static boolean isSlowSuccess(JobResult<?> result) {
        return result.isSuccess()
            && result.getExecutionTimeMsOrZero() > 60_000;
    }
}

scheduler.enqueue(() -> analyzeData())
    .when(DataAnalysisConditions::isSlowSuccess,
          () -> alertSlowExecution())
    .when(JobResult::isFailure,
          () -> notifyAdmins())
    .submit();

Custom workflow predicates are analyzed at submission time and stored as JobPayload JSON. Put compound logic in a public helper method or CDI bean method, then pass a method reference or a single-call lambda.

when (with priority)

<T> JobBuilder when(
    SerializablePredicate<JobResult<T>> condition,
    SerializableCheckedRunnable next,
    int priority)

Same as when() but with an explicit evaluation priority. Lower priority numbers are evaluated first, equal priorities preserve registration order, and only the first matching branch runs.

Parameters:

• condition -- predicate evaluated against the JobResult.
• next -- the task to execute when the condition is true.
• priority -- evaluation priority (lower = higher priority, default is 0).

scheduler.enqueue(() -> processOrder())
    .when(JobResult::isSuccess, () -> confirmOrder(), 0)    // evaluated first
    .when(JobResult::isFailure, () -> cancelOrder(), 1)     // evaluated second
    .submit();

whenResult

<T> JobBuilder whenResult(
    SerializableFunction<T, Boolean> condition,
    SerializableCheckedRunnable next)

Schedules a job based on the return value of the current job. The condition function receives only the value, not the full JobResult. As with when(), only the first matching branch runs; the scheduler cancels its siblings.

Type Parameters:

• T -- the type of the job's return value.

Parameters:

• condition -- function that evaluates the return value and returns a boolean.
• next -- the task to execute when the condition returns true.

public final class StockConditions {
    public static boolean isHighStock(Integer stock) {
        return stock > 100;
    }

    public static boolean isLowStock(Integer stock) {
        return stock > 0 && stock <= 100;
    }

    public static boolean isOutOfStock(Integer stock) {
        return stock == 0;
    }
}

scheduler.enqueue(() -> inventoryService.checkStock(itemId))
    .whenResult(StockConditions::isHighStock, () -> placeOrder(itemId))
    .whenResult(StockConditions::isLowStock, () -> alertLowStock(itemId))
    .whenResult(StockConditions::isOutOfStock, () -> markOutOfStock(itemId))
    .submit();

branch

JobBuilder branch(WorkflowCondition condition,
                  SerializableCheckedRunnable next,
                  String description)

Adds a workflow branch with an explicit WorkflowCondition and a human-readable description for monitoring and debugging. Branches are evaluated by lower-numbered priority first and then by registration order. The first match runs and all remaining siblings are canceled.

Parameters:

• condition -- the workflow condition determining when this branch fires.
• next -- the task to execute.
• description -- human-readable description for logs and dashboards.

public final class MetricsConditions {
    public static boolean isSlowJob(JobResult<?> result) {
        return result.getExecutionTimeMsOrZero() > 30_000;
    }
}

scheduler.enqueue(() -> analyzeMetrics())
    .branch(WorkflowCondition.custom(MetricsConditions::isSlowJob),
            () -> alertSlowJob(),
            "Alert ops when analysis takes over 30 seconds")
    .submit();

Submission

submit

JobHandle submit()

Submits the configured job (including the primary task, chain tasks, and workflow branches) to the scheduler for persistence and execution.

Returns: a JobHandle containing the unique job ID.

JobHandle handle = scheduler.enqueue(() -> processData())
    .withMaxRetries(3)
    .submit();

log.info("Submitted job {}", handle.id());

Accessor methods

These methods allow reading the configured state of a builder. They are primarily used internally by the scheduler but are part of the public interface.

Method	Return Type	Description
task()	SerializableCheckedRunnable	The primary task
workflowBranches()	List<WorkflowBranch>	Immutable list of workflow branches
opts()	JobOptions	Current immutable job options snapshot
params()	Map<String, String>	Unmodifiable parameter map
tags()	List<String>	Unmodifiable tag list
resourceName()	String	Resource name, or null if not set
isImmediate()	boolean	Whether immediate wakeup is requested
onSuccess()	SerializableConsumer<JobContext>	Success callback, or null

Example: complete configuration

JobHandle handle = scheduler.enqueue(() -> paymentService.charge(orderId, amount))
    // Execution options
    .withPriority(JobPriority.HIGH)
    .withMaxRetries(3)
    .withBackoff(BackoffPolicy.EXPONENTIAL, Duration.ofSeconds(5))
    .withTimeout(Duration.ofMinutes(2))
    // Resource limiting
    .withResource("payment-gateway")
    // Identity
    .withBusinessKey("charge-" + orderId)
    .withIdempotencyKey(requestId)
    // Metadata
    .withTags("payments", "orders")
    .withParam("orderId", String.valueOf(orderId))
    .withParam("amount", amount.toString())
    // Callbacks
    .onSuccess(ctx -> log.info("Payment charged for order {}", ctx.param("orderId")))
    .onFailure((ctx, err) -> alertService.paymentFailed(ctx.param("orderId"), err))
    // Workflow branches
    .thenOnSuccess(() -> fulfillmentService.startFulfillment(orderId))
    .thenOnFailure(() -> orderService.markPaymentFailed(orderId))
    // Immediate cluster notification
    .immediate()
    .submit();

See also

• JobSchedulerService Reference
• JobOptions Reference
• WorkflowCondition Reference
• Functional Interfaces