Java Workers

查看 Markdown查看 Markdown

This page shows how to start a Java Worker. It lists the Java names for the Worker settings and diagnostics. The Workers and the task queue page describes their behavior for every SDK.

Start a Worker

This example serves a Greeting Flow through a pull Worker. Its Step stores a greeting, and the greeting RPC reads it. The Worker allows eight concurrent tasks and keeps one slot for RPCs. See the Java SDK README for startup details.

public final class PullGreetingExample {
    private PullGreetingExample() {
    }

    public static RunningPullGreeting start(
            final BlobCache cache, final String serverAddress) throws InterruptedException, TimeoutException {
        final GreetingFlow flow = new GreetingFlow();
        final Registry registry = new Registry(List.<Flow<?>>of(flow));
        final Worker worker = new Worker(registry, cache, WorkerOptions.newBuilder()
                .serverAddress(serverAddress)
                .pull(PullWorkerOptions.newBuilder()
                        .maxConcurrentTasks(8)
                        // Keep one slot for RPCs so Flows stay responsive.
                        .reservedRpcTasks(1)
                        .build())
                .build());
        final Client client = new Client(registry, cache, new ClientOptions(serverAddress));
        final CompletableFuture<Void> workerResult = new CompletableFuture<>();
        final Thread workerThread = new Thread(() -> {
            try {
                worker.start();
                workerResult.complete(null);
            } catch (final Throwable failure) {
                workerResult.completeExceptionally(failure);
            }
        }, "pull-greeting-worker");
        workerThread.start();
        final RunningPullGreeting running = new RunningPullGreeting(
                flow, worker, client, workerThread, workerResult);
        try {
            worker.waitPullRegistered(Duration.ofSeconds(30));
        } catch (final InterruptedException | TimeoutException | RuntimeException | Error failure) {
            try {
                running.close();
            } catch (final InterruptedException | RuntimeException | Error cleanupFailure) {
                failure.addSuppressed(cleanupFailure);
                if (cleanupFailure instanceof InterruptedException) {
                    Thread.currentThread().interrupt();
                }
            }
            throw failure;
        }
        return running;
    }

    public static boolean isPullGreetingReady(final Worker worker) {
        return worker.getPullRegistrationState().isReady();
    }

    public static PullGreetingDiagnostics getPullGreetingDiagnostics(final Worker worker) {
        return new PullGreetingDiagnostics(
                worker.getPullRegistrationState().isReady(), 0L);
    }

    // reservedRpcDeclines is deprecated and always 0: reserved execution slots never decline tasks.
    public record PullGreetingDiagnostics(boolean isReady, long reservedRpcDeclines) {
    }

    public record RunningPullGreeting(
            GreetingFlow flow, Worker worker, Client client, Thread workerThread,
            CompletableFuture<Void> workerResult) implements AutoCloseable {
        @Override
        public void close() throws InterruptedException {
            client.close();
            worker.close();
            workerThread.join();
            workerResult.join();
        }
    }
}

Example: examples/java/src/main/java/io/superdurable/dex/production/pullworkers/PullGreetingExample.java

Readiness

Java provides waitPullRegistered(Duration) and getPullRegistrationState() in SDK 1.14.0. Java's stopOnPermanentStreamRejection provides the same opt-in shutdown behavior as Go.

A PullRegistrationError reports POLL_WORKER_SLOT_KIND_STEP_AND_RPC or POLL_WORKER_SLOT_KIND_RPC_ONLY as its slot kind.

Call the boolean helper from your probe endpoint and return 200 when true or 503 when false.

    public static boolean isPullGreetingReady(final Worker worker) {
        return worker.getPullRegistrationState().isReady();
    }

Example: examples/java/src/main/java/io/superdurable/dex/production/pullworkers/PullGreetingExample.java

Capacity and shutdown

Setting	Java name
Concurrent tasks	maxConcurrentTasks
Reserved RPC slots	reservedRpcTasks
Stop drain timeout	stopDrainTimeout
Shutdown	stop

The option names are builder methods.

The idle stream option, idleStreamsPerQueue, is deprecated. It is still accepted and has no effect.

Reserved RPC capacity

The reserve-decline counter, getPullReservedRpcDeclines(), is deprecated. It remains for source compatibility and always reads zero.

This diagnostics helper exposes registration readiness. Its reserve-decline field is always zero. A full Worker stays ready.

    public static PullGreetingDiagnostics getPullGreetingDiagnostics(final Worker worker) {
        return new PullGreetingDiagnostics(
                worker.getPullRegistrationState().isReady(), 0L);
    }

    public record PullGreetingDiagnostics(boolean isReady, long reservedRpcDeclines) {
    }

Example: examples/java/src/main/java/io/superdurable/dex/production/pullworkers/PullGreetingExample.java