---
title: "Run Java workers"
description: "Register business stages and let the SDK own transport, readiness, leases, delivery and shutdown."
category: "Java SDK"
---

A worker 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.

## Run the same handlers over either transport

This is the complete entry point from the [getting-started application](/docs/java/getting-started):

```java
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](/docs/java/configuration). HTTP is the default. Both transports use the same stages and durable server state.

| Transport | Configuration | Connection requirements |
| --- | --- | --- |
| HTTP | `transport: http` | Reach the Pipelogiq API. |
| AMQP | `transport: amqp` | Reach the API and the RabbitMQ address returned during bootstrap. |

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:

```java
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](/docs/java/workflows#share-contracts-with-an-independent-producer). 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:

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

For automatic constructor injection, use the [Spring Boot starter](/docs/java/spring-boot).

## 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](/docs/java/workflows#return-failures-that-express-intent) 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.

## Understand lease and result delivery

The runtime acquires work, maintains its lease and heartbeats, dispatches the handler, and reports context changes, logs and results. HTTP and AMQP have different delivery mechanics, but both require idempotent business effects because attempts can be delivered more than once.

A lease, result retry or broker acknowledgement is not a database transaction around your code. Use a stable business-action key or pipeline/stage identity for deduplication in your own database. An execution ID changes across attempts; a pipeline idempotency key covers pipeline creation, not every effect inside it.

## Distinguish started, ready and fully subscribed

The normal `run()` path performs startup and readiness checks. For an embedded worker, use `start()` and close its handle when the host stops. Inside a method using the getting-started classes:

```java
try (var worker = Pipelogiq.worker()
    .config("pipelogiq.yml")
    .stages(ValidateOrder.class, RecordOrder.class)
    .start()) {
  var created = worker.client().pipeline("Embedded order")
      .stage(ValidateOrder.class, new Order("embedded-1", new java.math.BigDecimal("5.00")))
      .stage(RecordOrder.class)
      .send();
  worker.client().pipelines()
      .waitForCompletion(created.id(), java.time.Duration.ofSeconds(60));
}
```

`start()` returns after readiness or throws. `build()` creates a handle without connecting; use it for host-managed lifecycle and configuration tests. `runner()` exposes transport-specific controls. A closed handle is terminal: create a new handle to restart.

## Shut down gracefully

Standalone applications use `run()` and its shutdown hook. Embedded applications close `WorkerHandle`; Spring applications 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 default API. `PipelogiqOptions`, `PipelogiqApiClient` and transport runners remain available for custom hosting. An injected API client remains caller-owned. See [resource ownership](/docs/java/reference#api-requests-futures-and-resource-ownership) and [OpenTelemetry](/docs/java/observability) before sharing clients or telemetry across runtimes.