Operate Java workers

A worker connects an application's registered handler names to executable Java code. HTTP and AMQP workers use the same handler, context and result contracts. Choose a transport based on your application's network access and operational environment.

Choose a transport

HTTP is the simplest first worker when an application can reach the API. It does not remove broker infrastructure from Pipelogiq itself. AMQP is appropriate when your environment supports direct broker connectivity and you want workers to consume through RabbitMQ.

Neither transport provides exactly-once external side effects. Follow the idempotency guidance regardless of transport.

Run the same handlers over either transport

Use the project and OrdersApp class from Get started with Java. Install the AMQP artifact from the current SDK checkout:

cd "$PIPELOGIQ_SDK_DIR"
./mvnw -pl sdk-amqp -am install -DskipTests

Add this dependency to the application's existing dependencies element; keep its HTTP dependency:

<dependency>
  <groupId>com.pipelogiq</groupId>
  <artifactId>pipelogiq-sdk-amqp</artifactId>
  <version>0.5.0</version>
</dependency>

These are current source-build coordinates. Maven Central availability is not assumed.

Create src/main/java/example/TransportWorker.java:

package example;

import io.pipelogiq.sdk.amqp.AmqpPipelineRunner;
import io.pipelogiq.sdk.core.PipelogiqOptions;
import io.pipelogiq.sdk.core.execution.StageResult;
import io.pipelogiq.sdk.core.execution.WorkerRunner;
import io.pipelogiq.sdk.http.HttpPipelineRunner;
import java.time.Duration;

public final class TransportWorker {
  public static void main(String[] args) {
    String transport = args.length == 0 ? "http" : args[0];
    var builder = PipelogiqOptions.builder()
        .apiUrl(System.getenv().getOrDefault("PIPELOGIQ_API_URL", "http://localhost:8081"))
        .apiKey(System.getenv("PIPELOGIQ_API_KEY"))
        .workerName("orders-java-" + transport)
        .environment("development")
        .maxConcurrency(4)
        .apiTimeout(Duration.ofSeconds(30))
        .drainGracePeriod(Duration.ofSeconds(30));
    String brokerOverride = System.getenv("PIPELOGIQ_RABBITMQ_URL");
    if (brokerOverride != null && !brokerOverride.isBlank()) {
      builder.rabbitMqUrl(brokerOverride);
    }
    var options = builder.build();
    WorkerRunner selected = switch (transport) {
      case "http" -> new HttpPipelineRunner(options);
      case "amqp" -> new AmqpPipelineRunner(options);
      default -> throw new IllegalArgumentException("Use http or amqp");
    };
    try (WorkerRunner worker = selected) {
      worker.register("ValidateOrder", OrdersApp.Order.class, OrdersApp.ValidateOrder::new);
      worker.registerInstance("RecordOrder", Void.class, (unused, context) -> {
        OrdersApp.Order order = context.get("validatedOrder", OrdersApp.Order.class);
        if (order == null) return StageResult.missingRequiredData("Validated order is missing");
        context.put("recorded", true);
        return StageResult.success("Recorded demo order " + order.orderId());
      });
      Runtime.getRuntime().addShutdownHook(new Thread(worker::close));
      var lifetime = worker.start();
      worker.awaitReady(Duration.ofSeconds(30));
      System.out.println(transport + " worker ready");
      lifetime.join();
    }
  }
}

Return to the application directory, rebuild, and run one worker:

"$PIPELOGIQ_SDK_DIR/mvnw" -q compile dependency:copy-dependencies
java -cp 'target/classes:target/dependency/*' example.TransportWorker amqp

Use http to select the gateway transport. Submit orders with the OrdersApp submit command from the first guide. Normally, broker settings come from bootstrap. Set PIPELOGIQ_RABBITMQ_URL only when you need an explicit broker address override, for example when container hostnames differ from the address reachable by the Java process.

The override uses an AMQP URI. Keep broker credentials in runtime configuration. Under Pipelogiq's convention, a URI ending in a lone / selects RabbitMQ's default / virtual host; explicit custom or encoded virtual hosts keep their meaning.

Register handlers before starting

register(name, Input.class, factory) creates a handler for each execution attempt. Constructor injection works naturally: the factory can create the handler with your application services. If that handler implements AutoCloseable, the SDK closes it after its invocation actually returns, including failures.

registerInstance(name, Input.class, handler) keeps a caller-owned instance. Multiple executions can call it concurrently. Use it for stateless handlers or handlers whose shared state is designed for concurrency. The SDK does not close caller-owned instances.

Registration is frozen while a worker runs. Register every handler before start(), including an agent runtime's six handlers. An explicit name gives deployments control over the wire contract; Java class renames should not silently rename a live handler.

Distinguish started, ready and fully subscribed

start() returns a CompletableFuture<Void> for the worker's lifetime. It is not a readiness signal. awaitReady(timeout) waits for successful transport initialization and an accepted online heartbeat. Observe both readiness and the lifetime future so a failed worker does not leave an apparently healthy application.

For AMQP, server-owned handler queues can appear only after the first dispatch. A worker may become ready while its heartbeat reports degraded coverage because some queues are missing. It retries those subscriptions automatically. Waiting for every lazy queue before submitting the first pipeline would create a deadlock.

Keep clientOwnedTopology at its default false when the server manages queues. clientOwnedTopology(true) permits compatible durable queue creation when your deployment explicitly uses that model. Incompatible declarations are configuration errors; repeatedly redeclaring a queue does not repair an argument mismatch.

Bound execution and external I/O

maxConcurrency limits active delivery processing. HTTP uses one long poll per handler queue; idle queues do not reserve an execution slot. Received deliveries wait for capacity, and the gateway may throttle further pulls.

Handlers execute synchronously on worker-managed threads. sendAsync() on a pipeline builder makes submission asynchronous; it does not turn StageHandler.execute into a future-returning interface. If your handler starts its own threads or executor tasks, it owns their lifetime and must explicitly propagate cancellation and trace context.

Give external HTTP/database calls timeouts and call context.throwIfCancelled() inside long loops. Thread interruption is cooperative. Java cannot safely kill arbitrary application code, and a handler that ignores cancellation can remain alive after its result is no longer publishable.

Understand lease and result delivery

For current execution deliveries, the worker acquires a stage lease and renews it while executing and reporting. If another delivery already owns or completed the execution, the rejected acquisition prevents duplicate handler execution for that fenced delivery.

After the handler returns, the worker reports the result before acknowledging the source delivery. A transient result outage retries the same serialized result while the lease remains valid, rather than rerunning the handler merely because reporting failed. AMQP requires successful mandatory routing and publisher confirmation before source acknowledgement.

Lease loss suppresses a late result. This protects accepted workflow state; it does not undo a network call or database commit the handler already performed. Timeouts return a classified TIMEOUT result and discard partial context changes from the timed-out invocation. External action deduplication remains an application responsibility.

HTTP session rejection triggers fresh bootstrap. Broker disconnects rebuild AMQP subscriptions from refreshed bootstrap configuration. Invalid credentials, unauthorized queues and topology conflicts need a configuration correction rather than an application-level retry loop around start().

Shut down gracefully

Use stop(gracePeriod) to stop intake and let active work finish within a bounded grace period. After the deadline, the worker requests cancellation. close() uses the configured drainGracePeriod, then releases the runner's resources. A stopped runner can be restarted; a closed runner cannot.

The examples use a JVM shutdown hook so Ctrl+C follows that lifecycle. In a service container, set the process termination grace period long enough for your chosen drain interval and resource cleanup. In Spring or another framework, let the framework manage start, awaitReady and close; no Spring dependency or automatic hosted-service adapter is required by the SDK.

A handler that has not returned still owns its invocation resources. Do not interpret completion of a shutdown deadline as evidence that its external side effects stopped or rolled back.

Bring your own HTTP client and tracing

Both runners can receive a PipelogiqApiClient constructed with a caller-supplied Java HttpClient. Use that to apply your TLS trust store, proxy and connection settings. The SDK does not replace the supplied client's settings or close caller-owned API clients.

Java carries W3C traceparent and tracestate, binds a trace scope on each invocation thread, and exposes execution IDs and counters. This is propagation, not an OpenTelemetry exporter. Add an application observability bridge when you need exported spans or metrics. Avoid logging entire builder payloads: creation payloads can include the API key, and context may contain sensitive data.

See the configuration reference for defaults and the workflow recipes for failure classification and stage timeouts.