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

[wu-sheng]

Add ThreadPolicy.ioBound(N) to library-batch-queue, and fix BatchQueue.shutdown() ordering

• If this is non-trivial feature, paste the links/URLs to the design doc. — no separate design doc; the rationale is in library-batch-queue/CLAUDE.md.

• Update the documentation to include this new feature. — library-batch-queue/CLAUDE.md gains an ioBound section and a Shutdown order section. No user-facing docs change: nothing under docs/ references BatchQueue except the changelogs, and no shipped queue adopts ioBound, so the thread-name table in changes-10.4.0.md stays accurate.

• Tests(including UT, IT, E2E) are added to verify the new feature. — 5 new ThreadPolicy tests, 4 new shutdown tests. 76 pass in the module.

• If it's UI related, attach the screenshots below. — n/a

• If this pull request closes/resolves/fixes an existing issue, replace the issue number. Closes #.

• Update the CHANGES log.

---

1. ThreadPolicy.ioBound(N) — new

Declares that a queue's consumers spend most of their time blocked, typically on I/O. Such a queue runs its drain loops on virtual threads where the runtime provides them (JDK 25+, via VirtualThreads.createScheduledExecutor) and on N platform threads otherwise.

The count is identical on both paths, so concurrency, batching, back-pressure, drop semantics and per-partition ordering are unaffected by the fallback — only the thread substrate changes. The purpose is narrow: avoid paying for N parked OS threads. It does not raise concurrency, because the drain loop is the task processor and nothing is handed off — consumer concurrency is min(threadCount, partitionCount) on both paths. An IO-bound queue therefore wants threads and partitions 1:1 with a small per-partition buffer.

There is deliberately no CPU-proportional form and no cpuCores(...).ioBound(): virtual threads are not preemptive, so a CPU-only task holds its carrier to completion and would starve the pool shared with GRPCServer and HTTPServer. The API cannot express that combination. L1/L2/TopN stay on cpuCores/fixed.

No existing queue uses the new policy, so this branch is inert until something opts in.

2. shutdown() ordering — pre-existing bug

shutdown() ran its final drain on the caller's thread while drain loops could still be inside consume(), invoking the same handler from two threads and breaking the single-drain-thread invariant that workers such as MetricsAggregateWorker document (its own lines 53-54). New order:

CAS shutdownStarted          at most one caller runs the sequence
running = false              reject produce(); drain chains stop re-queueing
rebalanceFuture.cancel(true) the periodic task does not read `running`
scheduler.shutdown()
awaitTermination(config)     wait for in-flight consume() to finish
final drain + dispatch       now uncontended
restore interrupt            only if interrupted, and only after dispatch

The order is load-bearing: the rebalance task must be cancelled before awaitTermination (on a virtual-thread scheduler its loop exits only on interrupt, and shutdown() does not interrupt), and running = false must precede the cancel because the rebalance fence exits via running, not the interrupt — LockSupport.parkNanos returns on interrupt without throwing.

The platform scheduler now also sets setExecuteExistingDelayedTasksAfterShutdownPolicy(false) so drain tasks parked on their idle backoff do not hold up termination.

On timeout the queue logs and proceeds rather than calling shutdownNow(): no consumer is written to tolerate interruption mid-batch, and a truncated executeBatch or stream send is worse than the race being fixed. shutdownTimeoutMs is per-queue (default 500ms).

Severity in practice is latent rather than live. shutdownAll() has no production callers; the only live path is GRPCRemoteClient.close(), whose consumer creates a fresh StreamObserver per batch and tolerates the concurrent call. The queues that would not tolerate it are never shut down. Correct by coincidence on both sides.

3. Fix a long-standing oal-rt test bug that has kept master's CI red

Four tests register scopes in @BeforeAll with a single try around the whole block:

try {
    listener.notify(Service.class);
    listener.notify(Endpoint.class);
    ...
} catch (RuntimeException e) {
    // Scopes may already be registered by other tests
}

notify() throws UnexpectedException when a scope id is already registered, so once any sibling
has registered even one scope, the first call throws and every scope after it is silently
skipped. The tests then fail at use time with ScopeDefine name = ... not found.

OALSourceAttributionTest registers Service alone. Whenever surefire's file-system run order
puts it first, OALClassGeneratorV2Test loses Endpoint, ServiceRelation, K8SServiceInstance
and TCPService — exactly the four scopes CI reports, in exactly the four failing test methods.
That ordering occurs on the CI runners but not on macOS, which is why it reproduces only in CI.

Reproduced locally with -Dsurefire.runOrder=reversealphabetical (5 tests, 4 errors, identical
messages). Each registration is now guarded independently, and the module passes under
filesystem, alphabetical and reversealphabetical.

This is a pre-existing bug, not one this branch introduced — master's own scheduled runs
(102af09, dae21ce, a2f1f16, a2498aa) fail the same way. It is folded in here because
fail-fast cancels the rest of the matrix, so nothing else can go green until it is fixed.

Reviewer notes

• The virtual-thread path IS covered by CI. The unit-test matrix includes JDK 25, so
  ThreadPolicy.ioBound takes the virtual branch there rather than the platform fallback.
• The once-only shutdown guard is covered through its completion-sharing behaviour
  (testLosingShutdownCallerAwaitsCompletion).

Each guard was verified by reverting the production change and confirming the corresponding test
fails: expected: <1> but was: <2> for concurrent dispatch, on both the ordering fix and the
timeout path, and expected: <0> but was: <1> for a losing shutdown caller returning early.

Out of scope

No change to any existing queue's configuration. Every current caller uses a non-ioBound policy,
so that branch is dead until something opts in.

The periodic rebalancer deliberately shares the drain pool rather than having its own thread: L1 is
cpuCores(1.0), one thread per core, and a dedicated rebalance thread would make it N+1 threads on
N cores — the oversubscription that costs context switches and cache reloads. Sharing also means
rebalancing is naturally deferred while all drain loops are busy, which is the right moment not to
be reshuffling partition ownership. This PR leaves that as is.