perf: reduce processing pool queue-lock contention with a sharded task queue (druid.processing.numThreadPools)

[rbankar7]

Fixes #19937.

Description

Under very high task rates, the processing pool's single task queue becomes a
lock-contention bottleneck. Every per-segment scan/merge task is submitted to a
single PrioritizedExecutorService backed by one PriorityBlockingQueue, which
—being a binary heap— is guarded by a single ReentrantLock for both put
and take (unlike LinkedBlockingQueue's two-lock design). So every task pays
the lock twice, and all numThreads workers plus all producers serialize on one
lock. On many-core nodes serving many small segments at high QPS, contention on
this one lock (AQS park/unpark + cache-line bouncing on the AQS state word)
dominates, showing up as high query wait time while CPU stays low — and it gets
worse as the node's thread count grows.

This PR adds an option to split the processing pool into N independent pools
("shards"), each with its own PriorityBlockingQueue and its own lock, so each
queue lock only sees ~1/N of the submit/take traffic and ~1/N of the contending
threads. It is fully opt-in and defaults to the existing single-pool behavior.

See the issue (#19937) for the motivation and profiling background.

Added druid.processing.numThreadPools (default 1)

druid.processing.numThreads is split as evenly as possible across the pools
(remainder to the first pools), so total thread count and processing-buffer
sizing are unchanged (direct-memory accounting is unaffected). The value is
clamped to [1, numThreads] (you can't have more pools than threads).
numThreadPools = 1 preserves the current behavior exactly.

Added ShardedPrioritizedExecutorService

A composite ListeningExecutorService holding N ordinary
PrioritizedExecutorService instances. Per-task work (execute/submit) is
routed to a shard chosen with ThreadLocalRandom — no shared counter, so the
router adds no contention of its own — and the chosen shard does all the
priority wrapping and ordering. Lifecycle calls fan out to every shard;
getQueueSize()/getActiveTasks() are summed across shards.

Design choice — routing: random routing was chosen over round-robin (which
needs a shared atomic counter, reintroducing a contended cache line) .
For a homogeneous, high-rate
task stream, random spreads load evenly enough that per-shard queue depths stay
balanced.

Trade-off: priority ordering becomes per-shard rather than global — a
high-priority task in one shard does not preempt work queued in another. For the
high-throughput, effectively-single-priority per-segment workload this pool
serves, that is an acceptable exchange; deployments needing strict global
priority ordering should keep the default (numThreadPools = 1).

Added ProcessingPoolStats interface

Extracted a small interface (getQueueSize() / getActiveTasks()) implemented
by both PrioritizedExecutorService and ShardedPrioritizedExecutorService.
MetricsEmittingQueryProcessingPool now emits segment/scan/pending and
segment/scan/active via instanceof ProcessingPoolStats instead of a concrete
class, so those metrics keep working for both pool types (summed across shards
for the sharded case) — no metric names change.

Wiring

DruidProcessingModule.createProcessingExecutorPool selects the sharded
implementation only when numThreadPools > 1; otherwise it uses the existing
PrioritizedExecutorService unchanged.

Release note

Added druid.processing.numThreads Pools (default 1), which splits the
processing pool into that many independent thread pools/queues, each with its own
lock. Increasing it relieves contention on the single processing-queue lock under
very high task rates (e.g. Historicals scanning many small segments at high QPS),
at the cost of per-pool rather than global task-priority ordering. The default
(1) is identical to previous behavior.

---

Key changed/added classes in this PR

• ProcessingPoolStats (new) — capability interface for pool queue/active-task stats.
• ShardedPrioritizedExecutorService (new) — composite of N PrioritizedExecutorService shards with random routing.
• PrioritizedExecutorService — implements ProcessingPoolStats; extracted a shared makeThreadPoolExecutor helper; added a startup log line.
• DruidProcessingConfig — new numThreadPools property (clamped to [1, numThreads]) + getter; backward-compatible constructor retained.
• MetricsEmittingQueryProcessingPool — emits scan metrics via ProcessingPoolStats.
• DruidProcessingModule — selects the sharded pool when numThreadPools > 1.

---

This PR has:

• been self-reviewed.
  • using the concurrency checklist.
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths.
• been tested in a test Druid cluster.