---
title: "Build Java workflows"
description: "Typed pipeline creation, context, retries, dependencies, idempotency, approval and dynamic stages."
category: "Java SDK"
---

Use a shared typed client to submit and manage pipelines. A producer describes durable work; workers execute its business stages. The [Java demo](/downloads/pipelogiq-java-demo.zip) contains every runnable recipe referenced here and needs only `pipelogiq-sdk`.

## Create a typed pipeline

With the files from [getting started](/docs/java/getting-started), the core operation is:

```java
import io.pipelogiq.sdk.Pipelogiq;
import java.math.BigDecimal;

// Inside an application method using the example Order and stage classes:
try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var created = client.pipeline("Process order")
      .idempotencyKey("order-123")
      .label("service", "orders")
      .context("source", "checkout")
      .stage(ValidateOrder.class, new Order("123", new BigDecimal("42.50")))
      .stage(RecordOrder.class)
      .send();
  System.out.println(created.id());
}
```

The annotation on each stage supplies its wire handler name. The default display name is the same name. Use `.stage("Validate payment", Handler.class, input)` when you need a distinct display name, including repeated uses of one handler. Dependency references use these submitted stage names.

A draft shares its client's connection and needs no separate closing. In a server application, inject or retain one client for the application lifetime rather than opening one per request.

## Share contracts with an independent producer

A producer does not need worker implementation classes. Share the agreed input schema and wire names:

```java
import io.pipelogiq.sdk.core.client.StageRef;

StageRef<Order> validate = StageRef.of("ValidateOrder", Order.class);
StageRef<Void> record = StageRef.noInput("RecordOrder");

try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var created = client.pipeline("External order")
      .idempotencyKey("external-order-123")
      .stage(validate, new Order("123", new java.math.BigDecimal("15.00")))
      .stage(record)
      .send();
  System.out.println(created.id());
}
```

This fragment uses `Pipelogiq` and `Order` from the previous example. `IndependentProducerExample.java` is a complete version with its own DTO and no imports of worker handlers:

```sh
java -cp 'target/classes:target/dependency/*' example.IndependentProducerExample external-001 15.00
```

Leave the orders worker running. Expect `COMPLETED` and `Recorded: true`. A contracts module can hold `StageRef` constants and records; another language can implement the same JSON contract.

## Understand inputs, context and labels

| Concept | Purpose | API |
| --- | --- | --- |
| Stage input | Explicit command for one stage | `.stage(Handler.class, input)` |
| Context | Execution data shared with subsequent stages | `.context`, `context.put`, `context.require` |
| Sensitive context | Data marked for protected API/display handling | `.sensitiveContext`, `context.putSensitive` |
| Labels | Classification and filtering | `.label(key, value)` |

Use `NoInputStageHandler` for a stage reading context only. There is no need for a dummy input or `Void.class` in its business method. Use `require` for a mandatory value and `get` for an optional one. Parallel branches should write distinct keys and join before consuming each other's results.

Labels are not an extra copy of context. Do not put secrets in labels or logs. Historical JSON property names may mention keywords for protocol compatibility; the public Java API uses labels and has no `withKeyword` alias.

## Observe and cancel a pipeline

Inside a method with an existing pipeline ID:

```java
import java.time.Duration;
import io.pipelogiq.sdk.core.client.PipelineStatus;

try (var client = Pipelogiq.connect("pipelogiq.yml")) {
  var snapshot = client.pipelines().get(123);
  System.out.println(snapshot.status());

  var terminal = client.pipelines().waitForCompletion(123, Duration.ofSeconds(60));
  if (terminal.status() == PipelineStatus.COMPLETED) {
    Boolean recorded = terminal.context("recorded", Boolean.class);
    System.out.println(recorded);
  }
  for (var stage : terminal.stages()) {
    System.out.println(stage.name() + ": " + stage.status() + " / " + stage.lastErrorCode());
  }
}
```

Replace `123` with your ID. The SDK polls with a bounded deadline and propagates errors. Terminal does not mean successful: a returned pipeline may be failed or cancelled. A wait timeout ends local observation, leaving the durable workflow unchanged.

`sendAsync`, `getAsync`, `waitForCompletionAsync` and other named async methods return `CompletableFuture`. Keep the client open while they run. Cancelling such a future stops that local request/wait. For server-side cancellation, call `client.pipelines().cancel(id)` or `cancelAsync(id)` explicitly.

## Return failures that express intent

Expected conditions should return a structured outcome:

| Result | Meaning |
| --- | --- |
| `StageResult.success()` | Successful stage; context changes are still returned |
| `validationError(message)` | Invalid data; terminal |
| `missingRequiredData(message)` | Required data absent; terminal |
| `businessRejected(message)` | Business rule refuses the operation; terminal |
| `upstreamError(message)` | Temporary upstream failure; retryable |
| `timeout(message)` | Deadline failure; retryable |
| `rateLimitExceeded(message)` | Rate limit; retryable |
| `transportUnavailable(message)` | Temporary connection/transport failure; retryable |
| `retryableError(message, code)` / `terminalError(message, code)` | Application-defined code and explicit classification |

Configure retry and deadline policy on the submitted stage. This is the policy used by the runnable inventory recipe:

```java
import io.pipelogiq.sdk.core.builders.StageOptions;
import io.pipelogiq.sdk.core.client.Backoff;
import java.time.Duration;

var retry = new StageOptions()
    .maxRetries(2)
    .retryInterval(Duration.ofSeconds(1))
    .backoff(Backoff.EXPONENTIAL)
    .maxRetryInterval(Duration.ofSeconds(5))
    .retryOnErrorCodes("UPSTREAM_ERROR")
    .timeout(Duration.ofSeconds(15));
```

Pass it to `.stage(FetchInventory.class, retry)` for a no-input stage, or `.stage(Handler.class, input, retry)` for a typed input. The server schedules retry attempts. The handler returns an outcome and does not sleep or implement a retry loop. Duration-based stage policies require whole seconds.

## Run the retry and dynamic-stage recipe

The downloaded `WorkflowExample.java` defines `FetchInventory`, `AppendAudit` and `AuditOrder`. Stop the basic worker, then start the extended worker in Terminal A:

```sh
java -cp 'target/classes:target/dependency/*' example.WorkflowExample worker
```

Terminal B:

```sh
java -cp 'target/classes:target/dependency/*' example.WorkflowExample submit workflow-001
```

The inventory stage deliberately returns `UPSTREAM_ERROR` on attempt one. A server-scheduled retry succeeds, the order is recorded, and an audit stage is appended. The final output is:

```text
Pipeline <id>: COMPLETED
Inventory available: true
Audit appended: true
Stages: 5
```

Use a fresh order ID to exercise a fresh retry. The example only updates workflow context; it does not call a real inventory service.

## Run a branch-and-join workflow

`StageOptions.dependsOn("stageName", ...)` and `runInParallelWith("stageName", ...)` describe server execution policy. They are not Java thread instructions. Give branches explicit unique stage names and reference those names, not Java class names or handler annotations. A join should depend on every branch whose context it needs.

For example, an order workflow can validate first, check stock and customer limits in parallel, then record only after both checks. Implement each branch as a separate stage, write separate context keys and test the graph against the server. Additional policies are `runNextIfFailed`, `failIfOutputEmpty`, `notifyOnFailure` and `runAsUser`; select them deliberately rather than bypassing a failed business prerequisite by default.

## Keep creation and side effects idempotent

Use a stable `.idempotencyKey(...)` for one business submission. The demo derives it from the order ID. A repeated request returns the same pipeline; a new business action needs a new key. If a create response is lost, use `client.pipelines().getByIdempotencyKey(key)` to reconcile before accidentally creating a second operation.

Pipeline creation idempotency does not make handler effects exactly once. For a database write, payment or notification, deduplicate using an action-specific business key or stable pipeline/stage identity in the receiving system. An execution ID changes on retry. The pipeline's key alone may be too broad when several stages perform different actions.

## Pause for a human decision

A handler can return `StageResult.waitingForApproval("Review the order")`. The stage waits instead of progressing automatically. An authenticated approval service can use the typed client:

```java
// Inside an authorized approval request, with the actual waiting stage ID:
client.stages().approve(stageId);
// Or, on the rejection path:
client.stages().reject(stageId, "Customer cancelled");
```

These are alternative actions, not two calls to execute together. Verify that the reviewer may decide this exact pipeline/stage. Sending a notification alone does not authorize a mutation. The [agent harness](/docs/java/testing) exercises waiting, approved and rejected paths without a provider.

## Append work dynamically

A producer can append typed definitions through `client.pipelines().append(pipelineId, StageDefinition.of(Handler.class, input))`.

Inside a handler, return dynamic stages with its result so the server applies them at the stage boundary. The current result API accepts the descriptor's JSON representation; you do not write wire fields yourself:

```java
return StageResult.success("Append audit")
    .append(StageDefinition.of(AuditOrder.class).toJson());
```

This fragment comes from `WorkflowExample.AppendAudit`; import `io.pipelogiq.sdk.core.client.StageDefinition`. The receiving worker must register `AuditOrder` before it starts. Typed pipeline management and ordinary creation do not need this conversion.

## Register and operate a schedule

Use `client.schedule(name)` with `.cron(...)`, `.every(...)` or `.runOnceAt(...)`, then `.apply()`. Schedule registration is an explicit producer operation, separate from worker startup. The [schedule guide](/docs/java/schedules) covers complete commands, time zones, policies, manual triggering and cleanup.