Java: Standardize hooks + middleware implementation with other languages

java/API.md

@@ -15,7 +15,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 | `JobUpdate` output parameter, `RecordOutput` | `Client.output`, `WorkContext.output` | Names describe the supported operation; attempt output is buffered until completion. |
 | `JobCompleteTx` | `Client.complete(connection, job)`, `WorkContext.complete(connection)` | Completion keeps `Job<A>` and commits with application changes. Only a running attempt can be completed. A committed completion remains authoritative if the handler subsequently throws. |
 | Context client, cancellation, cancel/snooze errors | `WorkContext.client`, cancellation methods, `cancel`, `discard`, `snooze` | Java has no Go context parameter convention. Control methods end the attempt; virtual-thread interruption supplements cooperative cancellation. |
-| Hooks and middleware | `Extension` | Default methods group related callbacks. JDBC hooks and middleware may throw checked exceptions; insert middleware receives a standard `Callable<T>`. |
+| Plugins, hooks, and middleware | `Plugin`, `HookInsertBegin`, `HookPeriodicJobsStart`, `HookWorkBegin`, `HookWorkEnd`, `JobInsertMiddleware`, `WorkerMiddleware` | `withPlugins` activates every hook and middleware interface on a value. `withHooks` and `withMiddleware` restrict registration to that side. Plugins run first, as in Go. Functional interfaces support lambdas and checked exceptions. |
 | Retry policy | `RetryPolicy` | A lambda receives the full job and returns `Duration`; the default uses the same error-count backoff and jitter. |
 | `Queues`, queue CRUD/control APIs | `Workers.addQueue/removeQueue`, `Client.queues()` | Runtime concurrency and persisted queue state have separate owners. Reads and mutations support caller transactions. |
 | `Subscribe`, `EventKind`, `Event` | `Workers.subscribe`, `EventKind`, `Event`, `Subscription` | Enum filters and an `AutoCloseable` subscription replace channels. Callbacks are synchronous; queue events include their queue. |
@@ -27,6 +27,14 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 
 ## Changes made before release
 
+- Replaced `Extension` and `withExtension` with separate hook and middleware
+  interfaces. Names follow Go, while functional interfaces allow Java lambdas.
+  Work-end hooks can preserve, replace, or clear the failure; a work-begin failure
+  skips the handler and work-end hooks. The conformance claim barrier now uses
+  worker middleware instead of an application-facing `afterClaim` callback.
+- Added the common `Plugin` marker and `withPlugins` registration to install all
+  of an implementation's hooks and middleware together. Renamed the internal Pro
+  storage interface and registration from `Plugin`/`withPlugin` to `Pilot`/`withPilot`.
 - Removed no-op `Client.close()`. Close `Workers` and the application-owned pool.
 - Added typed retrieval and typed transactional completion, including running-state validation.
 - Unified bulk insertion around typed homogeneous lists and mixed-kind submissions,
@@ -44,7 +52,7 @@ Java exposes every Go helper. Stored values and protocol behavior still follow G
 - Changed retry callbacks to receive the full job, enabling kind- and metadata-specific policies.
 - Added named migration options and fluent uniqueness state/kind overrides.
 - Kept destructive reset and retention helpers behind the internal driver seam.
-  `Plugin`, `Client.Driver`, protocol JSON helpers, and explicitly marked companion
+  `Pilot`, `Client.Driver`, protocol JSON helpers, and explicitly marked companion
   methods are not stable application extension APIs.
 
 Options copy collections and metadata on construction; `InsertOptions.metadata()`

java/README.md

@@ -285,7 +285,7 @@ requested. Call `context.checkCancelled()` when abandoning unfinished work.
 ### Getting the client within workers
 
 [`context.client()`](https://riverqueue.com/docs/context-client) gives a handler
-its client, including configured extensions. Use its transaction overloads when
+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.
 
 ```java
@@ -300,22 +300,21 @@ workerConfig.add(welcome, context ->
 Java exceptions and uncaught `Error`s are handled as failed attempts; River
 persists failures and retries according to policy. For
 [application error reporting](https://riverqueue.com/docs/error-handling), attach
-an `Extension.afterWork` hook. `Workers.Builder.errorHandler` separately receives
+a `HookWorkEnd` hook. `Workers.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.
 
 ```java
-var monitoredClient = client.withExtension(new Extension() {
-  @Override
-  public void afterWork(WorkContext<?> context, Throwable failure) {
-    if (failure != null) {
-      System.getLogger("jobs").log(System.Logger.Level.ERROR,
-          "Job " + context.job().id() + " failed", failure);
-    }
+HookWorkEnd reportFailure = (context, failure) -> {
+  if (failure != null) {
+    System.getLogger("jobs").log(System.Logger.Level.ERROR,
+        "Job " + context.job().id() + " failed", failure);
   }
-});
+  return failure;
+};
+var monitoredClient = client.withPlugins(reportFailure);
 var workerConfig = monitoredClient.workers()
     .queue("default", 10)
     .errorHandler(failure ->
@@ -326,6 +325,88 @@ var workerConfig = monitoredClient.workers()
 Hooks also see the control exceptions used for snoozing and cancellation. Keep
 reporting hooks fast and avoid throwing from them.
 
+### Plugins, hooks, and middleware
+
+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:
+
+```java
+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:
+
+```java
+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-persisted logging
 
 [Job logs](https://riverqueue.com/docs/job-logging) live in `river:log` metadata