Add ThreadPolicy.ioBound for blocking consumers, and fix BatchQueue shutdown ordering

docs/en/changes/changes.md

@@ -256,6 +256,7 @@
   admin-host only" entry above for the public REST retirement.
 * Drop six unused test-scoped dependencies from `runtime-rule` (`library-integration-test`, `library-banyandb-client`, `storage-banyandb-plugin`, `testcontainers`, `testcontainers:junit-jupiter`, `grpc-testing`). They staged the plugin-side ITs that were retired in favour of e2e; that coverage now lives in `test/e2e-v2/cases/runtime-rule/` (MAL over BanyanDB / PostgreSQL / Elasticsearch, LAL, meter, and the two-node cluster case). The module has no ITs today, and JUnit and Mockito are inherited from the root POM.
 * Declare `server-testing` at `test` scope everywhere. It ships only test scaffolding (`ModuleManagerTesting`, `MockModuleManager`, the MAL/LAL/Hierarchy rule loaders) plus two empty `org.junit` stubs that let Testcontainers' `GenericContainer` hierarchy resolve without JUnit 4, but four modules declared it at compile scope — including the `server-configuration` parent, so all eight `configuration-*` children inherited it — which put those `org.junit` stubs on the runtime classpath that `server-starter` copies into `oap-libs`. Modules whose tests need the stubs now declare the dependency themselves rather than inheriting it transitively, and `library-banyandb-client` gains the direct `library-util` dependency its `BanyanDBClient` always needed (it was resolving `StringUtil` through `server-testing`, a test-support module).
+* Add `ThreadPolicy.ioBound(N)` to `library-batch-queue`, for queues whose consumers spend most of their time blocked. Such a queue runs its drain loops on virtual threads where the runtime provides them (JDK 25+) and falls back to N platform threads otherwise; the count, and therefore concurrency, batching, back-pressure, drop semantics and per-partition ordering, are identical on both paths. Shutdown latency is the one exception: the platform scheduler drops drain tasks parked on their idle backoff, while the virtual-thread adapter sleeps inside the submitted task and cannot, so an `ioBound` queue should keep `maxIdleMs` within `shutdownTimeoutMs`. There is deliberately no CPU-proportional form: virtual threads are not preemptive, so CPU-bound work would hold its carrier and starve the shared carrier pool, and L1/L2/TopN stay on `cpuCores`/`fixed`. Also fixes `BatchQueue.shutdown()`, which ran its final drain on the caller's thread while drain loops could still be inside `consume()`, invoking a handler concurrently and breaking the single-drain-thread invariant workers such as `MetricsAggregateWorker` rely on: it now cancels the periodic rebalance task, waits for in-flight consumers (`shutdownTimeoutMs`, default 500ms per queue), and serialises its final dispatch behind a read/write dispatch lock so the guarantee holds even when that wait times out or is interrupted. Drain loops hold the read lock for the whole cycle — the running recheck, the partition dequeue, the idle notification and the dispatch — because `onIdle()` touches the same worker state as `consume()` and an unlocked dequeue would let shutdown dispatch a newer batch ahead of one a task already holds. Concurrent shutdown callers await the winner's completion rather than returning early. A consumer is never interrupted mid-batch.
 
 #### OAP Server
 * Fix LAL's `segmentId` and `spanId` extractor statements, which the grammar accepted and the parser never implemented. `LALParser.g4` declares `traceIdStatement`, `segmentIdStatement` and `spanIdStatement`, and the codegen already carried `setSegmentId`/`setSpanId` in its setter table, but `LALScriptParser.visitExtractorStatement` had a branch for only the first of the three. The remaining alternatives fell through to a line that assumed whatever was left had to be an `ifStatement`, so a rule writing `segmentId ...` failed at boot with a `NullPointerException` naming `IfStatementContext` — for a rule line containing no `if`. Both statements now work, and an unhandled extractor statement reports its own rule line instead of throwing. Existing log records are unaffected: `LogBuilder` copies trace id, segment id and span id straight from the log's metadata, and only skips that copy when a rule has set them — which no shipped rule did, which is why the gap went unnoticed. Dedicated execution tests now cover reading all three fields from `log.traceContext.*` and writing all three from an extractor.

oap-server/oal-rt/src/test/java/org/apache/skywalking/oal/v2/generator/MetricDefinitionEnricherTest.java

@@ -53,16 +53,24 @@ public class MetricDefinitionEnricherTest {
 
     @BeforeAll
     public static void initializeScopes() {
-        try {
-            DefaultScopeDefine.Listener listener = new DefaultScopeDefine.Listener();
-            listener.notify(Service.class);
-            listener.notify(Endpoint.class);
-            listener.notify(ServiceRelation.class);
-            listener.notify(K8SService.class);
-            listener.notify(K8SServiceInstance.class);
-            listener.notify(TCPService.class);
-        } catch (RuntimeException e) {
-            // Scopes may already be registered by other tests
+        // Each registration is guarded separately. notify() throws when a scope id is already
+        // registered by another test in this JVM, and a single try around the whole block would
+        // swallow that and silently skip every scope after it — leaving the rest unregistered and
+        // failing later with "ScopeDefine name = ... not found". Which scope trips first depends
+        // on surefire's run order, so the truncation only shows up on some platforms.
+        final DefaultScopeDefine.Listener listener = new DefaultScopeDefine.Listener();
+        for (final Class<?> scope : List.of(
+            Service.class,
+            Endpoint.class,
+            ServiceRelation.class,
+            K8SService.class,
+            K8SServiceInstance.class,
+            TCPService.class)) {
+            try {
+                listener.notify(scope);
+            } catch (RuntimeException e) {
+                // Already registered by a sibling test; the remaining scopes still need registering.
+            }
         }
     }
 

oap-server/oal-rt/src/test/java/org/apache/skywalking/oal/v2/generator/OALClassGeneratorV2Test.java

@@ -72,16 +72,24 @@ public class OALClassGeneratorV2Test {
 
     @BeforeAll
     public static void initializeScopes() {
-        try {
-            DefaultScopeDefine.Listener listener = new DefaultScopeDefine.Listener();
-            listener.notify(Service.class);
-            listener.notify(Endpoint.class);
-            listener.notify(ServiceRelation.class);
-            listener.notify(K8SService.class);
-            listener.notify(K8SServiceInstance.class);
-            listener.notify(TCPService.class);
-        } catch (RuntimeException e) {
-            // Scopes may already be registered by other tests
+        // Each registration is guarded separately. notify() throws when a scope id is already
+        // registered by another test in this JVM, and a single try around the whole block would
+        // swallow that and silently skip every scope after it — leaving the rest unregistered and
+        // failing later with "ScopeDefine name = ... not found". Which scope trips first depends
+        // on surefire's run order, so the truncation only shows up on some platforms.
+        final DefaultScopeDefine.Listener listener = new DefaultScopeDefine.Listener();
+        for (final Class<?> scope : List.of(
+            Service.class,
+            Endpoint.class,
+            ServiceRelation.class,
+            K8SService.class,
+            K8SServiceInstance.class,
+            TCPService.class)) {
+            try {
+                listener.notify(scope);
+            } catch (RuntimeException e) {
+                // Already registered by a sibling test; the remaining scopes still need registering.
+            }
         }
     }
 

oap-server/server-library/library-batch-queue/CLAUDE.md

@@ -48,7 +48,7 @@ use `BatchQueueManager.create(name, config)` which throws on duplicate names.
 | `BatchQueue<T>` | The queue itself. Holds partitions, runs drain loops, dispatches to consumers/handlers. |
 | `BatchQueueManager` | Global registry. Creates/retrieves queues by name. `create()` for unique, `getOrCreate()` for shared. |
 | `BatchQueueConfig<T>` | Builder for queue configuration (threads, partitions, buffer, strategy, consumer, balancer). |
-| `ThreadPolicy` | Resolves thread count: `fixed(N)`, `cpuCores(mult)`, `cpuCoresWithBase(base, mult)`. |
+| `ThreadPolicy` | Resolves thread count: `fixed(N)`, `cpuCores(mult)`, `cpuCoresWithBase(base, mult)`, `ioBound(N)`. |
 | `PartitionPolicy` | Resolves partition count: `fixed(N)`, `threadMultiply(N)`, `adaptive()`. |
 | `PartitionSelector<T>` | Routes items to partitions. Default `typeHash()` groups by class. |
 | `HandlerConsumer<T>` | Callback for processing a batch. Has optional `onIdle()` for flush-on-idle. |
@@ -60,13 +60,59 @@ use `BatchQueueManager.create(name, config)` which throws on duplicate names.
 ## ThreadPolicy
 
 ```java
-ThreadPolicy.fixed(4)              // exactly 4 threads
-ThreadPolicy.cpuCores(1.0)         // 1 thread per CPU core
+ThreadPolicy.fixed(4)                   // exactly 4 threads
+ThreadPolicy.cpuCores(1.0)              // 1 thread per CPU core
 ThreadPolicy.cpuCoresWithBase(1, 0.25)  // 1 + 0.25 * cores (e.g., 3 on 8-core)
+ThreadPolicy.ioBound(50)                // 50 threads whose consumers block; virtual when available
 ```
 
 Always resolves to >= 1.
 
+### ioBound — blocking consumers
+
+`ioBound(n)` declares that this queue's consumers spend most of their time blocked, typically on
+I/O. The scheduler then uses virtual threads where the runtime provides them (JDK 25+, via
+`VirtualThreads.createScheduledExecutor`) and falls back to `n` platform threads otherwise.
+
+**The count is identical on both paths.** Concurrency, batching, back-pressure, drop semantics and
+per-partition ordering are unaffected by the fallback. The queue logs once at creation when it
+asked for virtual threads and got platform ones.
+
+One behaviour does differ: **shutdown latency**. On the platform path the scheduler drops drain
+tasks still waiting out their idle backoff, so termination is immediate. The virtual-thread adapter
+implements a delay by sleeping *inside* the submitted task, which `shutdown()` cannot cancel, so
+termination waits out the longest outstanding backoff — bounded by `maxIdleMs`. Keep
+`maxIdleMs <= shutdownTimeoutMs` on an `ioBound` queue, or `shutdown()` will log a spurious
+"drain loops did not finish" warning on every teardown. Nothing is lost either way: a late-waking
+drain task sees `running == false` and exits without draining, and the final drain flushes the data.
+
+Sizing: the drain loop **is** the task processor — a consumer blocking for seconds occupies its
+loop for that whole time, and nothing is handed off. So consumer concurrency is
+`min(threadCount, partitionCount)`, and an IO-bound queue wants **threads and partitions 1:1**
+with a small per-partition buffer:
+
+```java
+.threads(ThreadPolicy.ioBound(concurrency))
+.partitions(PartitionPolicy.fixed(concurrency))   // 1:1, or the count is clamped away
+.bufferSize(4)                                    // per partition
+.strategy(BufferStrategy.IF_POSSIBLE)
+.minIdleMs(50).maxIdleMs(500)                     // seconds-long work; 1ms polling is waste
+```
+
+The count expresses **how many concurrent blocking calls the downstream service tolerates**, which
+does not follow core count — hence no CPU-proportional variant, and hence the count is mandatory.
+
+**Never use `ioBound` for CPU-bound work.** Virtual threads are not preemptive: a CPU-only task
+holds its carrier to completion, and the carrier pool is shared process-wide with the executors
+`GRPCServer` and `HTTPServer` already create. L1 (`MetricsAggregateWorker`), L2
+(`MetricsPersistentMinWorker`) and TopN do in-memory merges across hundreds of partitions and must
+stay on `cpuCores`/`fixed`. The API has no `cpuCores(...).ioBound()` form so this combination
+cannot be expressed.
+
+`ioBound` with a `DrainBalancer` logs a warning: rebalancing exists to redistribute skewed
+CPU-bound partitions, which is not this shape. It is not rejected — it remains safe — but it is
+usually a configuration mistake.
+
 ## PartitionPolicy
 
 ```java
@@ -221,5 +267,73 @@ mode:       single consumer (JDBC batch flush)
 3. `queue.addHandler(type, handler)` -- registers type handler (adaptive: may grow partitions)
 4. `queue.produce(data)` -- routes to partition, blocks or drops per strategy
 5. Drain loops run continuously, dispatching batches to consumers/handlers
-6. `BatchQueueManager.shutdown(name)` -- stops drain, final flush
-7. `BatchQueueManager.shutdownAll()` -- called during OAP server shutdown
+6. `BatchQueueManager.shutdown(name)` -- stops drain, waits for in-flight consumers, final flush
+7. `BatchQueueManager.shutdownAll()` -- available, but currently has no production callers
+
+### Shutdown order
+
+```
+CAS shutdownStarted          winner runs the sequence; losers await its completion latch
+running = false              reject produce(); drain chains stop re-queueing
+rebalanceFuture.cancel(true) the periodic task does not read `running`
+scheduler.shutdown()
+awaitTermination(config)     COURTESY wait for orderly exit; may time out or be interrupted
+dispatchLock.writeLock()     THIS is the guarantee — waits out any in-flight consume()
+final drain + dispatch
+restore interrupt            only if interrupted, and only after dispatch
+```
+
+The order is load-bearing:
+
+- **cancel before awaitTermination** -- on a virtual-thread scheduler the periodic task is one
+  submission whose loop exits only on interrupt, and `shutdown()` does not interrupt. Without the
+  cancel, termination can never be observed. (A `ScheduledThreadPoolExecutor` cancels periodic
+  tasks itself, so the platform path worked by relying on that default.)
+- **`running = false` before cancel** -- the rebalance fence spins on
+  `cycleCount.get(t) <= snap && running`, and `LockSupport.parkNanos` *returns* on interrupt
+  without throwing, leaving the flag set. The fence exits via `running`, not the interrupt.
+- **wait before the final drain** -- draining on the caller's thread while a drain loop is still
+  inside `consume()` invokes the same handler concurrently, breaking the single-drain-thread
+  invariant workers such as `MetricsAggregateWorker` depend on.
+
+The final drain deliberately ignores partition ownership, so partitions left `UNOWNED` by an
+interrupted rebalance are still flushed.
+
+**`awaitTermination` is not the safety mechanism -- the dispatch lock is.** The wait is a courtesy
+for orderly exit and can time out or be interrupted, and on either path a drain loop may still be
+running. `shutdown()` takes the WRITE lock, so its final dispatch waits the in-flight consumer out
+instead of racing it.
+
+Drain loops hold the READ lock for the **whole cycle** -- the `running` recheck, the partition
+dequeue, `notifyIdle()` and the dispatch -- not merely around `dispatch()`. All four are required:
+a dequeue outside the lock lets shutdown drain and dispatch a *newer* batch while a task holds an
+older one, which is then dispatched out of order after `shutdown()` has returned; `notifyIdle()`
+outside it runs `onIdle()` concurrently with the final dispatch, and implementations such as
+`MetricsAggregateWorker` flush the same worker state from `onIdle()`; and the loop's `running` test
+can pass just as shutdown completes. Read locks are shared, so drain loops still run concurrently
+with one another -- correct, since the partition selector routes a type to one partition and
+therefore one task. Nothing sleeps inside the body (backoff is applied by `scheduleDrain` after the
+loop exits), so the lock is held only for as long as the consumer runs. Blocking there is bounded by that consumer's
+own work and is preferable to interrupting it: no consumer is written to tolerate interruption
+mid-batch, and a truncated `executeBatch` or stream send is worse than the race being fixed, which
+is why `shutdownNow()` is never called.
+
+`shutdownTimeoutMs` is per-queue (default 500ms) and bounds only the courtesy wait; `0` skips it
+entirely, which is safe now that the lock -- not the wait -- provides the guarantee.
+
+Two further properties, both for the same reason -- a consumer must never be entered twice at once,
+nor with the interrupt flag set:
+
+- **Once-only, and shared completion.** Only the caller that wins a CAS runs the sequence; the
+  losers block on a completion latch rather than returning early, so `shutdownAll()` racing a
+  `shutdown(name)` cannot report completion while the winner is still draining. That wait is
+  deliberately **unbounded**: the winner may spend all of `shutdownTimeoutMs` in `awaitTermination`
+  and then block on the write lock for as long as a consumer runs, so any fixed bound would let a
+  loser return early. Interruption is deferred and restored on the way out.
+- **Interrupt is deferred.** If `awaitTermination` is interrupted, the queue logs, still runs the
+  final drain (skipping it would lose data), and re-asserts the thread's interrupt flag only after
+  `dispatch()` returns.
+
+The platform scheduler additionally sets `setExecuteExistingDelayedTasksAfterShutdownPolicy(false)`
+so drain tasks parked on their idle backoff do not hold up termination. Dropping them is safe: such
+a task, had it run, would find `running == false` and exit at the top of its loop without draining.

oap-server/server-library/library-batch-queue/pom.xml

@@ -28,6 +28,11 @@
     <artifactId>library-batch-queue</artifactId>
 
     <dependencies>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>library-util</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility</artifactId>