Java 21 client and worker runtime for River's Postgres and SQLite schemas. Jobs are ordinary River jobs: another language can insert, cancel, retry, or work them using the same database.

Java 21 or newer is required. Add com.riverqueue:river:0.4.1 and a JDBC driver to your application's dependencies. Jackson 3 is included transitively.

<repositories>
  <repository>
    <id>river</id>
    <url>https://riverqueue.com/maven</url>
    <releases><checksumPolicy>fail</checksumPolicy></releases>
    <snapshots><enabled>false</enabled></snapshots>
  </repository>
</repositories>

<dependency>
  <groupId>com.riverqueue</groupId>
  <artifactId>river</artifactId>
  <version>0.4.1</version>
</dependency>
<dependency>
  <groupId>org.postgresql</groupId>
  <artifactId>postgresql</artifactId>
  <version>42.7.12</version>
</dependency>

The River repository is public and requires no credentials. Maven Central supplies third-party dependencies.

For SQLite, replace the Postgres JDBC dependency with org.xerial:sqlite-jdbc:3.53.4.0. To install a local copy of River from source, see the development guide.

Define job arguments as a record, give the job a stable kind, and register a worker lambda. This complete Example.java starts a client, inserts a job, waits for its completion, and stops the workers. Set DATABASE_URL to a Postgres URL or a file-backed SQLite JDBC URL.

import com.riverqueue.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class Example {
  record SendEmail(String address, String subject) {}

  public static void main(String[] args) throws Exception {
    var database = Database.connect(System.getenv("DATABASE_URL"));
    new Migrator(database).migrate();

    var workers = new Workers()
        .add(JobType.of("send_email", SendEmail.class), context -> {
          System.out.println("Sending " + context.args().subject() + " to " + context.args().address());
          context.output("sent");
        });
    var client = Client.builder(database).queue(Client.QUEUE_DEFAULT, 10).workers(workers).build();
    var completed = new CompletableFuture<Client.Event>();
    try (var subscription = client.subscribe(completed::complete, Client.EventKind.JOB_COMPLETED);
        var run = client.start()) {
      client.insert(JobType.of("send_email", SendEmail.class),
          new SendEmail("hello@example.com", "Welcome"));
      System.out.println("Completed job " + completed.get(30, TimeUnit.SECONDS).job().id());
    }
  }
}

Run this demonstration against a development database with no other workers. The worker prints a message; replace its body with your mail service. In a service, keep the RunHandle open until the application stops. Each active job runs on a virtual thread; queue limits bound concurrent jobs. In production, run migrations as a deployment step using the CLI below.

Job kinds are explicit, stable wire names independent of Java class names. Record properties use snake_case in JSON by default, so accountId becomes account_id. Match kinds and JSON fields across languages sharing a job.

Use a pooled DataSource in applications. Database.connect(url) is a small DriverManager convenience for scripts and tests and opens a physical connection for each operation. River never closes an application-owned data source. An insert-only client is simply a Client instance without a worker runtime; inserting needs no start or stop call.

SQLite connections accept sqlite://path and sqlite:path, as well as native jdbc:sqlite: URLs. See the shared SQLite URL reference for relative paths and in-memory databases.

The CLI runs with Java 21 or newer. JBang resolves the thin CLI JAR and its dependencies and installs a reusable command:

jbang app install --name river-java --java 21 \
  --java-options=--enable-native-access=ALL-UNNAMED \
  --repos https://riverqueue.com/maven --repos central \
  --main com.riverqueue.cli.MigrationCli \
  com.riverqueue:river-cli:0.4.1
export PATH="$HOME/.jbang/bin:$PATH"
export DATABASE_URL=postgres://localhost/myapp
# For SQLite: export DATABASE_URL=jdbc:sqlite:/absolute/path/myapp.db

river-java --help
river-java migrate-list
river-java migrate-up --dry-run
river-java migrate-up

Dependencies, including both JDBC drivers, are downloaded once and cached. No all classifier is needed. For containers or installations with Maven and a JDK, use the Maven-only instructions below, or build from source. --database-url URL overrides DATABASE_URL. --schema background_jobs selects a Postgres schema; an actual migration up creates it if needed. Listing and dry runs do not create schemas or migration tables.

Use --target-version N to stop at a version, --max-steps N to limit the count, and --dry-run --show-sql to inspect a planned change. On migrate-down, target version 0 explicitly removes the entire migration line. A zero --max-steps uses the command's default: all pending up migrations or one down migration.

river-java migrate-down --max-steps 2 --dry-run --show-sql
river-java migrate-up --target-version 8
river-java migrate-get --version 3 --up > river_3.up.sql
river-java migrate-get --driver sqlite --all --up > river_sqlite.up.sql

SQL export defaults to Postgres without a database URL. Use --driver sqlite for SQLite, or select the dialect through the URL. Export accepts comma-separated --version values or --all; --exclude-version 1 excludes migration history setup when exporting for another migration framework. Exit codes are 0 for success, 1 for an operation failure, and 2 for invalid arguments.

The migration history is shared with River clients in other languages. Use the CLI version matching your Java library. The Java API remains available for embedded use:

var result = new Migrator(database).migrate();
System.out.println("Applied migrations: " + result.applied());
var preview = new Migrator(database).migrate(Migrator.Direction.DOWN,
    Migrator.Options.defaults().maxSteps(2).dryRun(true));

In an empty directory, create this consumer POM and download the runtime JARs:

cat > pom.xml <<'EOF'
<project>
  <modelVersion>4.0.0</modelVersion>
  <groupId>local</groupId><artifactId>river-cli-install</artifactId><version>1</version>
  <repositories><repository><id>river</id><url>https://riverqueue.com/maven</url></repository></repositories>
  <dependencies><dependency><groupId>com.riverqueue</groupId><artifactId>river-cli</artifactId><version>0.4.1</version></dependency></dependencies>
</project>
EOF
mvn --batch-mode --no-transfer-progress \
  org.apache.maven.plugins:maven-dependency-plugin:3.11.0:copy-dependencies \
  -DincludeScope=runtime
java --enable-native-access=ALL-UNNAMED -cp 'target/dependency/*' \
  com.riverqueue.cli.MigrationCli --help

Replace --help with any migration command above. Copy the entire target/dependency directory into a deployment image and adjust the classpath to its location. Running it needs only Java; Maven and network access are needed when downloading the dependencies.

Enqueue in the application's transaction so the job and application writes commit or roll back together. Pass a JDBC connection with auto-commit disabled. River uses savepoints and never commits, rolls back the outer transaction, or closes that connection. Insert hooks and middleware run inside the same transaction.

try (var connection = applicationDataSource.getConnection()) {
  connection.setAutoCommit(false);
  try {
    accounts.create(connection, account);
    client.insert(connection, email, new SendEmail(account.email(), "Welcome"));
    connection.commit();
  } catch (Exception failure) {
    connection.rollback();
    throw failure;
  }
}

client.transaction(connection -> ...) manages a new transaction for River-owned operations. client.transaction(existingConnection, connection -> ...) groups multiple operations under one savepoint in an existing transaction. Neither Client nor Database needs closing; the application owns its pool. Workers should be idempotent: process failure can cause an attempt to run again after its external effects have succeeded.

Bulk insertion inserts a list atomically. For an existing JDBC transaction, use the connection-first overload of either form below. Homogeneous batches take a common job type. Mixed batches use submissions with their own kind and options. Both return insertion results containing the persisted Job<JsonNode> values.

var inserted = client.insertMany(email, List.of(
    new SendEmail("one@example.com", "Welcome"),
    new SendEmail("two@example.com", "Welcome")));

record GenerateReport(long accountId) {}
var report = JobType.of("generate_report", GenerateReport.class);
var mixed = client.insertMany(List.of(
    email.submission(new SendEmail("one@example.com", "Welcome")),
    report.submission(new GenerateReport(42), InsertOptions.builder().queue("reports").build())));

Job operations live on client.jobs(), alongside queue operations on client.queues(). Use jobs().get, list, cancel, delete, deleteMany, retry, complete, and update. Insertions remain client.insert and client.insertMany. Connection-first overloads use an existing transaction; jobs().complete always requires one.

client.jobs().get(id) returns Job<JsonNode> because an ID does not identify a Java argument class. Supply a job type when it is known:

Job<SendEmail> job = client.jobs().get(jobId, email);
System.out.println(job.args().address());

Typed retrieval checks the kind and decodes the arguments. context.complete(connection) also retains its Job<A> argument type. Insertion results contain Job<JsonNode> because uniqueness can return an existing job with another kind or argument schema. The returned arguments are the persisted values, including any additional fields written by another language. See the API review for the full application surface and prerelease API changes.

Lists default to ascending job ID order. Keep the filters and order unchanged when continuing from a page's cursor:

var query = JobQuery.builder().kinds("send_email").limit(50).build();
var first = client.jobs().list(query);
if (!first.jobs().isEmpty()) {
  var next = client.jobs().list(query.after(first.cursor()));
}

Use order(JobQuery.Order.TIME) for ordering by the timestamp appropriate to the selected state, or SCHEDULED_AT or FINALIZED_AT for an explicit timestamp.

A thrown exception triggers retries until maxAttempts is exhausted, when the job becomes discarded. The default policy uses River's quartic backoff with jitter. Configure a different policy on the client builder, or explicitly retry an existing job with client.jobs().retry(id). An explicit retry that updates a job notifies its queue after commit. Use client.jobs().retry(connection, id) to make the retry part of an existing transaction. A retry policy receives the full job snapshot before the current failure is appended to errors; use job.errors().size() + 1 to include that failure. It runs only when another attempt is possible. Return a nonnegative Duration; if the policy throws, returns null or a negative delay, or overflows the retry time, River reports the problem and uses its default backoff.

var retryingEmail = email.withDefaults(InsertOptions.builder().maxAttempts(10).build());
var clientBuilder = Client.builder(database)
    .queue(Client.QUEUE_DEFAULT, 10)
    .retryPolicy(job -> Duration.ofSeconds(Math.min(300, 5L * (job.errors().size() + 1))))
    .worker(retryingEmail, context ->
        mailer.send(context.args().address(), context.args().subject()));
client.insert(retryingEmail, new SendEmail("hello@example.com", "Welcome"));

Call clientBuilder.build() to create the configured client, then call its start() method and keep the returned handle open as in the first example. Job-type defaults apply when inserting through that JobType; another producer must set its own compatible options.

These sections follow the non-Pro entries in the River documentation's Features sidebar. Snippets are independent and reuse client and database from the quick start, with imports from com.riverqueue, java.time, and java.util. Handler examples configure a builder; call build() after registering your handlers, then start that client. In runtime examples, run is the RunHandle returned by client.start(). mailer, accounts, and deliveries are application services, applicationDataSource is your JDBC connection pool, application.awaitStop() waits for your service to stop, and jobId is an existing job's ID.

var email = JobType.of("send_email", SendEmail.class);
var clientBuilder = Client.builder(database).queue(Client.QUEUE_DEFAULT, 10);

Cancel an enqueued or running job with client.jobs().cancel(id). Running attempts receive a cooperative cancellation signal. A handler can cancel itself permanently with context.cancel(reason); this ends the attempt without further retries.

client.jobs().cancel(jobId);

clientBuilder.worker(email, context -> {
  context.checkCancelled();
  if (context.args().address().isBlank()) context.cancel("Missing address");
  mailer.send(context.args().address(), context.args().subject());
});

Long-running handlers should check cancellation between operations and use interruptible I/O. context.awaitCancellation() waits for the signal. A normal return reports success even if cancellation or a timeout was requested. Call context.checkCancelled() when abandoning unfinished work.

Client.Builder.jobTimeout sets the default attempt timeout. Override Handler.timeout(job) for a particular worker; the callback receives typed arguments and can choose a duration per job:

clientBuilder.worker(email, new Workers.Handler<SendEmail>() {
  @Override
  public Duration timeout(Job<SendEmail> job) {
    return Duration.ofMinutes(5);
  }

  @Override
  public void work(WorkContext<SendEmail> context) {
    mailer.send(context.args().address(), context.args().subject());
  }
});

Returning null or Duration.ZERO inherits the runtime default, as lambda handlers do. A negative duration disables the timeout. The callback also runs when the leader considers abandoned jobs for rescue, so keep it thread-safe and free of side effects. Rescue respects longer worker deadlines and disabled timeouts.

context.client() gives a handler its client, including configured hooks and middleware. Use its transaction overloads when enqueueing follow-up work that must commit with other writes or job completion.

record Welcome(String address) {}
var welcome = JobType.of("welcome", Welcome.class);
clientBuilder.worker(welcome, context ->
    context.client().insert(email, new SendEmail(context.args().address(), "Welcome")));

Java exceptions and uncaught Errors are handled as failed attempts; River persists failures and retries according to policy. For application error reporting, attach a HookWorkEnd hook. Client.Builder.errorHandler separately receives runtime failures, such as database or subscriber errors. Failures in observers or event subscribers are reported without changing the job's outcome or preventing other callbacks from running. If the runtime error handler itself throws, River logs both failures and continues.

HookWorkEnd reportFailure = (context, failure) -> {
  if (failure != null) {
    System.getLogger("jobs").log(System.Logger.Level.ERROR,
        "Job " + context.job().id() + " failed", failure);
  }
  return failure;
};
var clientBuilder = Client.builder(database).plugins(reportFailure)
    .queue(Client.QUEUE_DEFAULT, 10)
    .errorHandler(failure ->
        System.getLogger("river").log(System.Logger.Level.ERROR, "Runtime failure", failure))
    .worker(email, context -> mailer.send(context.args().address(), context.args().subject()));

Hooks also see the control exceptions used for snoozing and cancellation. Keep reporting hooks fast and avoid throwing from them.

Use client.withPlugins(...) to install hooks and middleware together. Plugin is their common marker interface: Hook and Middleware both extend it. A plugin automatically activates every operation-specific interface it implements.

Hooks observe individual lifecycle stages through HookInsertBegin, HookWorkBegin, HookWorkEnd, and HookPeriodicJobsStart. Middleware wraps an operation with a next callback through JobInsertMiddleware and WorkerMiddleware. Each is a functional interface for lambdas or method references. For example, individual callbacks can be installed together:

JobInsertMiddleware logInsert = (connection, submissions, next) -> {
  var results = next.call();
  System.getLogger("jobs").log(System.Logger.Level.DEBUG,
      "Inserted " + results.size() + " jobs");
  return results;
};
WorkerMiddleware timeWork = (context, next) -> {
  long started = System.nanoTime();
  try {
    next.run();
  } finally {
    System.getLogger("jobs").log(System.Logger.Level.DEBUG,
        "Job " + context.job().id() + " took " + (System.nanoTime() - started) + " ns");
  }
};
var instrumentedClient = client.withPlugins(reportFailure, logInsert, timeWork);

A reusable plugin can implement multiple interfaces on one class:

final class JobInstrumentation implements HookWorkEnd, WorkerMiddleware {
  @Override
  public void work(WorkContext<?> context, WorkContext.Step next) throws Exception {
    System.getLogger("jobs").log(System.Logger.Level.DEBUG,
        "Starting job " + context.job().id());
    next.run();
  }

  @Override
  public Throwable workEnd(WorkContext<?> context, Throwable failure) {
    if (failure != null) {
      System.getLogger("jobs").log(System.Logger.Level.ERROR,
          "Job " + context.job().id() + " failed", failure);
    }
    return failure;
  }
}

var instrumentedClient = client.withPlugins(new JobInstrumentation());

Use withHooks(...) or withMiddleware(...) when only that side of an implementation should be activated. All three registration methods return a new client and append to their respective lists. Use the returned client for insertion and worker configuration. Instances can run concurrently and must protect shared mutable state.

As in Go, plugins run first, followed by separately registered hooks or middleware, regardless of which registration method was called first. Each list preserves its registration order. Repeated registrations are retained, so registering the same instance twice invokes it twice for each applicable phase.

Hooks run in registration order inside the innermost middleware. Middleware runs in registration order before next and unwinds in reverse order afterward. Insertion middleware runs once per batch, including a one-element batch for a single insertion; insert hooks run once per job. Middleware receives an immutable list of submissions with the caller's arguments and option overrides. Insert hooks and middleware run in the insertion's transaction, and a failure anywhere rolls back the insertion, even after next returns.

Work hooks and middleware receive decoded arguments. A work-begin failure skips the handler and work-end hooks. Work-end hooks receive the handler's exception or Error, or null for success, and return the outcome passed to the next hook. Return the received failure to preserve it, a different failure to replace it, or null to report success. A thrown exception skips remaining work-end hooks. The internal Pilot interface is a River Pro storage integration point, separate from application plugins.

Job logs live in river:log metadata as an array of {attempt, log} objects. Java has no bundled job-logging middleware yet. You can explicitly write that shared format; this example appends a fixed message and retains the last ten attempt entries.

clientBuilder.worker(email, context -> {
  var logs = new ArrayList<Object>();
  context.job().metadata().path("river:log").forEach(logs::add);
  try {
    mailer.send(context.args().address(), context.args().subject());
  } finally {
    logs.add(Map.of("attempt", context.job().attempt(), "log", "Email attempt finished\n"));
    context.metadata("river:log", logs.subList(Math.max(0, logs.size() - 10), logs.size()));
  }
});

Metadata is saved when the attempt finishes, including on failure. General-purpose logging integrations should also bound each entry's byte size; this snippet does not capture SLF4J or System.Logger output automatically.

Queues isolate concurrency for different workloads. Set a queue at insertion and configure its worker count. The count is local to each runtime; it is not a cluster-wide concurrency limit.

Client.QUEUE_DEFAULT is the default queue name, "default". Jobs use it when neither insertion options nor job-type defaults specify a queue.

var clientBuilder = Client.builder(database)
    .queue(Client.QUEUE_DEFAULT, 10)
    .queue("mail", 4)
    .worker(email, context -> mailer.send(context.args().address(), context.args().subject()));
client.insert(email, new SendEmail("hello@example.com", "Welcome"),
    InsertOptions.builder().queue("mail").build());

A running handle also supports run.addQueue(name, count) and run.removeQueue(name), which drains that queue's active attempts. Use fetchOnlyKnownKinds(true) when sharing queues with other language workers that handle different kinds.

Pausing stops new fetches across clients sharing the database; active attempts may finish. Inserts continue while a queue is paused. Queues must already exist, for example after starting workers.

client.queues().pause("mail");
// Resume after the maintenance window.
client.queues().resume("mail");

Both operations also accept a JDBC transaction as their first argument.

Register periodic jobs on every potential leader with matching schedules and arguments. Only the elected leader enqueues them. The final boolean enables insertion whenever that client becomes leader. Five-field cron expressions support explicit time zones. Failures evaluating or inserting an individual periodic job are sent to the runtime error handler; other periodic jobs and maintenance continue running.

Cron schedules without a CRON_TZ= or TZ= prefix use the JVM's system time zone when run by workers, including daylight-saving changes. Use an explicit zone when leaders may run in different time zones. Direct schedule.next(...) calls use the reference time's zone or fixed offset unless the expression specifies one; pass a ZonedDateTime to retain a named zone's daylight-saving rules. Time-zone rules come from the JVM; schedules computed by River clients in other languages match only when they use the same time-zone data.

var clientBuilder = Client.builder(database)
    .queue(Client.QUEUE_DEFAULT, 10)
    .worker(email, context -> mailer.send(context.args().address(), context.args().subject()))
    .periodic("hourly-status", Schedule.every(Duration.ofHours(1)), email,
        new SendEmail("ops@example.com", "Hourly status"), InsertOptions.defaults(), true)
    .periodic("daily-report", Schedule.cron("CRON_TZ=America/Chicago 0 9 * * *"), email,
        new SendEmail("ops@example.com", "Daily report"), InsertOptions.defaults(), false);

Periodic IDs use the same syntax as job kinds and must fit within 127 UTF-8 bytes. Use null for an anonymous registration. run.periodicJobs().add(...) returns a handle for removing that registration. Builder registration returns no handle, so an anonymous definition added through the builder can only be removed with run.periodicJobs().clear().

OSS schedules are kept in memory and can miss occurrences during downtime or leader changes. Add or remove definitions on a running runtime:

var periodicJobs = run.periodicJobs(); // PeriodicJobBundle
var handle = periodicJobs.add(new PeriodicJob<>(
    "hourly-status", Schedule.every(Duration.ofHours(1)), email,
    new SendEmail("ops@example.com", "Hourly status"), InsertOptions.defaults(), true));
periodicJobs.remove(handle);
// Or remove a named registration without its handle:
periodicJobs.removeById("hourly-status");

Registrations belong to that runtime; configure them on every eligible leader. Only the leader inserts jobs. runOnStart also applies when adding a definition while leading. Removing a definition leaves previously inserted jobs in place, and an insertion already in progress may finish. An ID can be reused after removal. Periodic jobs require leader election to be enabled.

Definitions without runOnStart are picked up on the next maintenance pass. After stopping a runtime, its bundle still accepts additions and removals, but those changes do not enqueue jobs or restart the runtime.

The bundle also provides addMany, removeMany, removeManyById, and clear. addMany validates the entire batch before registering any jobs and returns handles in input order. Handles identify one registration, so removing an old handle cannot remove a replacement that reused the same ID. The same PeriodicJob definition can be passed to Client.Builder.periodic before start.

Interval schedules retain their planned cadence when maintenance runs late. Each job's scheduled_at defaults to its planned occurrence; an explicit scheduledAt option takes precedence.

Output is a JSON value stored under metadata.output. Set it in a worker with context.output; other languages and River UI can read the result once the attempt finishes.

clientBuilder.worker(email, context -> {
  mailer.send(context.args().address(), context.args().subject());
  context.output(Map.of("delivered", true, "address", context.args().address()));
});

var output = client.jobs().get(jobId).metadata().path("output");

To persist output immediately, use client.jobs().update with JobUpdateOptions:

var updates = JobUpdateOptions.builder()
    .metadata(Map.of("provider_request_id", requestId))
    .output(Map.of("normalized_text", normalized))
    .build();
var updated = client.jobs().update(jobId, updates);

// Or include the update in an existing JDBC transaction:
client.jobs().update(connection, jobId, updates);

Metadata merges at the top level, preserving unrelated keys. Supplied nested objects replace whole values; JSON null is preserved. Explicit output overrides any "output" in the metadata updates. Omit it to leave output unchanged, or call .output(null) to store JSON null. JobUpdateOptions.defaults() changes nothing. Missing jobs throw RiverException with code NOT_FOUND.

Explicit output in JobUpdateOptions and context.output is limited to 32 MiB of encoded UTF-8 JSON, including JSON escaping. Larger values throw IllegalArgumentException without changing previously recorded output.

The connection overload requires auto-commit disabled and leaves commit/rollback to the caller. Save step results before marking a resumable step complete. context.output remains buffered until the attempt finishes and may overwrite an immediate output update. Read fresh output with client.jobs().get(jobId); existing job objects are snapshots.

Named steps let retries skip work completed before an earlier failure. Keep step names and ordering stable across deployments and languages; anything outside a step runs on every attempt.

clientBuilder.worker(email, context -> {
  context.step("send", () ->
      mailer.send(context.args().address(), context.args().subject()));
  context.step("record-delivery", () -> deliveries.record(context.job().id()));
});

For loops, context.stepWithCursor(name, cursor -> ...) receives a saved JSON cursor, and context.cursor(value) updates it. Progress is normally buffered until the attempt ends. A process crash can lose buffered progress, so steps must remain idempotent.

Use context.checkpoint(connection) inside a step to persist its progress in the same transaction as application writes:

clientBuilder.worker(email, context -> {
  context.step("record-delivery", () -> context.transaction(connection -> {
    deliveries.record(connection, context.job().id());
    context.checkpoint(connection);
    return null;
  }));
});

Call context.checkpoint(connection, cursorValue) inside stepWithCursor to save cursor progress. A retry resumes that cursor step; a checkpoint without a cursor skips a completed ordinary step. The connection must have auto-commit disabled. Let transaction failures propagate out of the step so a rollback leaves the last committed checkpoint authoritative. Neither overload commits the caller's transaction.

Commit the transaction before the step callback returns successfully, and propagate transaction and commit errors from that callback. A successful callback marks the step complete even if an enclosing transaction later rolls back.

Schedule a job by supplying an Instant. The leader's scheduler makes it available once due; execution also depends on scheduler cadence and queue capacity.

client.insert(email, new SendEmail("hello@example.com", "Reminder"),
    InsertOptions.builder()
        .scheduledAt(Instant.now().plus(Duration.ofHours(3)))
        .build());

Snooze when work should wait without consuming a retry. context.snooze ends the current attempt and schedules the same job to run again; code after the call does not execute on that attempt.

clientBuilder.worker(email, context -> {
  if (!mailer.isReady()) context.snooze(Duration.ofMinutes(5));
  mailer.send(context.args().address(), context.args().subject());
});

Subscribe to events on a Client. Subscribe before start() to include events from immediately processed jobs. Subscriptions survive restarts of that client until closed. Callbacks are process-local, run synchronously, and should finish quickly. They are not a durable stream of all events in the cluster. Close the subscription to unregister it.

try (var subscription = client.subscribe(
    event -> System.out.println("Completed job " + event.job().id()),
    Client.EventKind.JOB_COMPLETED);
    var run = client.start()) {
  application.awaitStop();
}

Event kinds are Client.EventKind values. Omit the filter to receive every kind. Job events carry event.job(); queue events carry event.queue() and a null job. The enum's value() retains River's lower-case protocol spelling.

Use isolated database tests with JUnit and signals instead of sleeps. This complete test method uses a fresh SQLite file, subscribes before starting, and waits for committed completion. Add the SQLite JDBC driver and JUnit Jupiter to the test classpath.

@org.junit.jupiter.api.Test
void completes(@org.junit.jupiter.api.io.TempDir java.nio.file.Path directory) throws Exception {
  record Echo(String message) {}
  var echo = JobType.of("echo", Echo.class);
  var database = Database.connect("jdbc:sqlite:" + directory.resolve("river.db"));
  new Migrator(database).migrate();
  var client = Client.builder(database)
      .queue(Client.QUEUE_DEFAULT, 1)
      .worker(echo, context -> context.output(context.args().message()))
      .build();
  var completed = new java.util.concurrent.CompletableFuture<Client.Event>();

  try (var subscription = client.subscribe(completed::complete, Client.EventKind.JOB_COMPLETED);
      var run = client.start()) {
    var inserted = client.insert(echo, new Echo("hello"));
    var event = completed.get(5, java.util.concurrent.TimeUnit.SECONDS);
    org.junit.jupiter.api.Assertions.assertEquals(inserted.job().id(), event.job().id());
    org.junit.jupiter.api.Assertions.assertEquals(Job.State.COMPLETED, event.job().state());
    org.junit.jupiter.api.Assertions.assertEquals("hello", event.job().metadata().path("output").asString());
  }
}

Client.builder(database).clock(clock) and new Client(database, clock) accept an injected Clock for deterministic insert timestamps. The worker runtime uses wall time. Java does not currently provide a dedicated worker-test harness.

Complete a job atomically with application writes using context.complete(connection). If the transaction rolls back, neither change survives. Return from the handler after committing; external effects such as sending email cannot participate in a JDBC transaction.

clientBuilder.worker(email, context -> {
  context.transaction(connection -> {
    deliveries.record(connection, context.job().id());
    context.output(Map.of("recorded", true));
    context.complete(connection);
    return null;
  });
});

Uniqueness can combine arguments, queue, time period, and job states. A duplicate insert returns the existing job with uniqueSkippedAsDuplicate() set. The key format is shared with River clients in other languages.

var options = InsertOptions.builder()
    .unique(Unique.args().perQueue().per(Duration.ofHours(1)))
    .build();
var inserted = client.insert(email, new SendEmail("hello@example.com", "Welcome"), options);
boolean duplicate = inserted.uniqueSkippedAsDuplicate();

A period is an aligned time bucket, not a sliding delay since the last insertion. Use email.uniqueBy("address") to hash only selected top-level JSON fields, and insert with that returned JobType. Nested fields use component lists, such as email.uniqueBy(List.of(List.of("account", "id"))). Default unique states include completed jobs, so a retained completed row can still prevent insertion.

Workers can be functions. Java's Workers.Handler<A> accepts lambdas and method references, keeping arguments, registration, and behavior together without a worker subclass for every kind.

Workers.Handler<SendEmail> sendEmail = context ->
    mailer.send(context.args().address(), context.args().subject());
clientBuilder.worker(email, sendEmail);

Use a file database shared by every process. Connections opened by River enable WAL, foreign keys, and a five-second busy timeout. Configure application-created connections equivalently.

var database = Database.connect("jdbc:sqlite:/var/lib/myapp/river.db");
new Migrator(database).migrate();
var client = new Client(database);

SQLite permits one writer at a time: keep application transactions short, and insert using that transaction's connection. Do not open another writing connection while holding an application write transaction. In-memory databases with independent connections are not suitable for this runtime.

For Postgres, select an alternate schema on the Database used by both migrations and clients. All collaborating clients must target the same schema. The migrator creates it if needed.

var jobsDatabase = database.withSchema("background_jobs");
new Migrator(jobsDatabase).migrate();
var jobsClient = new Client(jobsDatabase);

Workers is a handler registry; it has no lifecycle methods. Client.start() returns the RunHandle that owns one execution. Only one run can be active on a client, including while a timed-out stop is draining. After stop() succeeds, that client can be started again with the same client ID and original settings. An old handle never stops a later run. Dynamic queue and periodic registrations belong to their run and are reset from client configuration on restart.

Registries and builders are mutable configuration objects. workers(registry) copies registrations, and build() snapshots settings. client.toBuilder() creates configuration for an independent client; withHooks, withMiddleware, and withPlugins likewise return new clients. Configure them before starting, and use the configured client for same-process insert wakeups.

For graceful stopping, run.stop() stops fetching and waits for active attempts. close() does the same, making try-with-resources convenient. run.stopAndCancel() also requests cancellation of active attempts. Both wait for handlers to return. A graceful-stop timeout leaves attempts running; call stop() again to keep waiting or stopAndCancel() to request cancellation. Attempt timeouts continue to apply during graceful stop. Once a handler returns, River retains its worker slot until completion is committed, retrying transient database failures even if the handler left its thread interrupted.

A normal handler return reports success, including after a shutdown signal. If a handler exits with work unfinished, call context.checkCancelled() or propagate InterruptedException; shutdown interruption requeues the job without consuming an attempt. awaitCancellation() only waits for the signal.

var client = Client.builder(database)
    .queue(Client.QUEUE_DEFAULT, 10)
    .stopTimeout(Duration.ofSeconds(30))
    .worker(email, context -> mailer.send(context.args().address(), context.args().subject()))
    .build();
var run = client.start();
try {
  application.awaitStop();
} finally {
  run.stop();
}

The timeout bounds each stop wait and a queue-removal drain; a timeout throws a RiverException. Cancellation is cooperative, followed by interruption after the configured stuck threshold. The JVM cannot forcibly terminate code that ignores interrupts; handlers must cooperate with cancellation.

Leader election coordinates Java and other River clients through the database. The leader runs periodic insertion and maintenance: scheduling due jobs, rescuing abandoned attempts, cleaning expired jobs and queues, and reindexing configured Postgres indexes. Leadership is enabled by default.

rescueAfter must not be shorter than jobTimeout, so another client does not rescue an attempt that is still allowed to run. Its default is one hour, plus an explicitly configured positive job timeout. Worker overrides are checked for each rescue candidate even when they exceed rescueAfter. Worker durations and snooze delays must fit in signed 64-bit nanoseconds, up to about 292 years.

var clientBuilder = Client.builder(database)
    .queue(Client.QUEUE_DEFAULT, 10)
    .retention(Duration.ofDays(1), Duration.ofDays(1), Duration.ofDays(7))
    .worker(email, context -> mailer.send(context.args().address(), context.args().subject()));

Retention arguments are cancelled, completed, and discarded durations, in that order. Negative durations retain that state indefinitely. If a runtime uses leadership(false), another eligible client must run maintenance; periodic registrations require leadership. run.isLeader() reports the local status. Leadership leases last serviceInterval plus ten seconds, keeping the same renewal margin even when the interval is increased. serviceInterval also controls maintenance and queue heartbeats. It must be positive and shorter than one day, the retention period for inactive queues. Short retries and snoozes, including rescued retries, use timed fetch wake-ups when they are due before the next maintenance interval. Rescued jobs also notify their queue's peers when due.

See developing River for Java.