Run Java workers

A worker is a separate, continuously running service. It needs configuration, stage classes and an entry point. The SDK owns the API client, transport connection, input conversion, heartbeats, leases, execution limits and result delivery. Applications that create pipelines use a producer client and do not start a worker in the same process.

Run the same handlers over either transport

This is the complete entry point from the getting-started application:

package example;

import io.pipelogiq.sdk.Pipelogiq;

public final class OrdersExample {
  public static void main(String[] args) {
    Pipelogiq.worker()
        .config("pipelogiq.yml")
        .stages(ValidateOrder.class, RecordOrder.class)
        .run();
  }
}

run() blocks until shutdown. It creates the runtime, registers handlers, connects, waits for readiness and installs a JVM shutdown hook. Ctrl+C or SIGTERM stops intake and drains active work within the configured grace period.

Choose a transport

Set pipelogiq.transport to http or amqp in configuration. HTTP is the default. Both transports use the same stages and durable server state.

The umbrella pipelogiq-sdk dependency includes both transports. AMQP normally obtains credentials and topology from the server; an amqp.url override is for network layouts where the returned broker address is not reachable. A connection failure is reported instead of silently changing transports.

Register handlers before starting

Implement StageHandler<T> for a typed input or NoInputStageHandler when a stage reads only execution context:

package example;

import io.pipelogiq.sdk.core.execution.*;

@Stage("RecordOrder")
public final class RecordOrder implements NoInputStageHandler {
  @Override
  public StageResult execute(StageContext context) {
    Order order = context.require("validatedOrder", Order.class);
    context.put("recorded", true);
    return StageResult.success("Recorded demo order " + order.orderId());
  }
}

@Stage("RecordOrder") fixes the handler's wire name. Producers can use the same class or a shared StageRef contract. Without the annotation, the class's simple name becomes its wire name; renaming it changes how work is routed.

The SDK infers input types from StageHandler<T>, including inherited generic types and List<Order>. Raw handlers and unresolved type variables fail registration. Duplicate names fail rather than replacing a handler silently.

Simple .stages(...) registration requires an accessible no-argument constructor and creates a fresh handler for each attempt. Constructor injection is available through a factory. This fragment goes in an application method where orderService and your ValidateOrderWithService implementation already exist:

Pipelogiq.worker()
    .config("pipelogiq.yml")
    .configure(worker -> worker.register(
        ValidateOrderWithService.class,
        () -> new ValidateOrderWithService(orderService)))
    .run();

For automatic constructor injection, use the Spring Boot starter.

Read and update context

context.require("key", Type.class) is for required data. Missing or null data produces MISSING_REQUIRED_DATA; incompatible present data produces VALIDATION_ERROR. context.get(...) is for optional data. Generic values can use Jackson's TypeReference<List<Order>> overload.

Use put, putSensitive and remove to change context. The SDK returns only changes made by this attempt. Context keys are case-insensitive; parallel branches should write distinct keys. Sensitive values remain available to authorized execution code, so do not include them in logs or labels.

context.logInfo, logWarning and logError attach messages to the execution. pipelineId(), stageId(), attempt() and executionId() identify the attempt. context.pipelogiq() exposes the worker's shared typed client; do not close that borrowed client inside a handler.

Bound execution and external I/O

max-concurrency limits concurrent stage executions. A stage's StageOptions.timeout(...) controls its execution deadline; api-timeout controls SDK API requests. Business HTTP calls and database queries also need their own deadlines.

Cancellation is cooperative. Call context.throwIfCancelled() before expensive work and between iterations, and honor thread interruption. Java cannot safely terminate arbitrary user code. The runtime fences late results, but cannot undo an external side effect that has already happened.

Return an explicit failure result for an expected business condition. The SDK converts unexpected exceptions into failed stage outcomes. Retrying a stage is server policy; do not implement an unbounded retry loop inside a handler.

Make external effects safe to repeat

The SDK automatically acquires and renews a lease before it calls a stage handler. The lease prevents two workers from reporting the same execution as current work. You do not configure or manage it in application code.

A lease cannot make an external side effect exactly once. Consider a payment stage:

1. The worker charges the customer.
2. Its connection fails before the result reaches Pipelogiq.
3. The server cannot confirm completion and dispatches another attempt.
4. Without deduplication, the second attempt can charge the customer again.

Pass a stable business key such as charge-order-<orderId> to the payment provider. For a database effect, store a stable action key under a unique constraint and perform the deduplication marker and business write in one database transaction. A pipeline idempotency key prevents duplicate pipeline creation; it does not protect every effect performed inside the pipeline. executionId() changes on every attempt, so it must not be the deduplication key. A business identifier is usually best; pipelineId() plus stageId() is available when the effect is unique to that stage.

HTTP and AMQP differ in delivery and acknowledgement, but this rule is the same for both. Result retries and message acknowledgements are SDK concerns; business idempotency is an application concern.

Keep the worker online

Deploy the worker as its own service and let run() own its lifecycle. It blocks after a successful startup, keeps reconnecting after transient transport failures, and continues accepting work until the process receives Ctrl+C or SIGTERM. Run more than one replica when you need capacity or availability; leases coordinate competing workers.

The producer runs separately:

• an API service can create a pipeline in response to a request;
• a scheduled job can create or manage schedules;
• a CLI can submit and inspect work;
• none of them registers handlers or starts a worker.

The dashboard reports the worker online after bootstrap, transport setup and an accepted heartbeat. With AMQP, server-owned stage queues may appear after the first dispatch; the worker subscribes automatically and reports a temporarily degraded state while a subscription is missing. Application code does not wait for or manage these subscriptions.

Shut down gracefully

Plain Java workers use run() and its shutdown hook. Spring worker services let the application context close the worker. Shutdown stops new intake and allows active executions to finish up to drain-grace-period, then requests cancellation.

Advanced registration accepts StageScope.owned(handler), StageScope.scoped(handler, resourceScope) or StageScope.borrowed(handler). An owned handler is closed; a scoped registration closes the supplied scope; a borrowed registration closes neither. registerInstance(...) shares the instance across attempts and requires thread-safe implementation.

Bring your own HTTP client and tracing

The typed facade is the application API. PipelogiqOptions, PipelogiqApiClient, WorkerHandle and transport runners remain available for framework adapters such as the Spring starter and for SDK-level integration tests. They are not required to run a worker service. See resource ownership and OpenTelemetry for advanced integrations.