[improve][client] Coalesce message listener drain scheduling

[lhotari]

Motivation

Each consumer message-listener notification currently allocates and submits a drain task to the internal executor. One task can drain multiple messages, so bursts also queue redundant tasks and executor queue nodes.

Modifications

• Add a per-consumer ListenerTaskScheduler that reuses one drain task and coalesces notifications with atomic idle/scheduled/triggered-again states.
• Requeue a follow-up drain at the executor tail when another notification arrives. The corresponding message-arrival task may be queued behind the current drain, so immediately draining again or simply discarding that notification could leave a message waiting indefinitely.
• Reset scheduling state in finally, including after drain failures. Propagate submission rejection and reset state so a later notification can retry. Delivery is not guaranteed while the executor rejects work.
• Keep the existing message-drain body and per-message application-listener dispatch on their existing executors. The scheduler requires the internal executor's asynchronous, serial FIFO execution.
• Add deterministic scheduler tests, a concurrent arrival stress test, and a JMH comparison of fresh tasks, reused tasks and coalescing.

The atomic reset separates the races: a trigger before the reset marks the state for a follow-up; a trigger after it schedules a task itself or merges with the follow-up. A race may cause an extra empty drain, but does not discard a notification on an accepting executor.

Verifying this change

The added tests cover arrivals queued behind a drain, burst coalescing, return to idle, notifications during draining, exceptions with and without pending notifications, rejection of initial and follow-up submissions, and five concurrent runs of four producers with 5,000 arrivals each. Existing consumer close/seek, listener ordering, partitioned and Shared/Key_Shared subscription, and custom listener-executor tests are also included in scoped validation.

Historical E020 measurements (not a new benchmark run for this extraction):

Notifications per JMH turn	Fresh tasks, ns/turn	Coalesced, ns/turn	Fresh / coalesced B/turn
1	99.900	109.935	88 / 64
16	1411.001	883.019	1408 / 576
1024	88600.559	47592.202	90113 / 32833

JMH used Corretto 25, ZGC with a 1 GiB heap, two forks, three 1-second warmups, five 1-second measurements and the GC profiler. The benchmark models queued arrivals and drains with a real LinkedBlockingQueue; it excludes OS thread handoff, decoding and application callbacks. Coalescing improves bursts but adds about 10 ns for an isolated notification.

A historical four-Shared-consumer profiling comparison estimated consumer allocation at 11.264 -> 10.498 GB (-6.8%) and executor queue-node allocation at 1.917 -> 1.487 GB (-22.4%). Throughput and CPU samples were essentially flat; zero failed ACKs were reported. These are sampled allocation estimates from single integration runs, with allocation-site variability and host temperatures reaching 100 C, so they do not establish a general throughput or CPU improvement. Raw profiles and adjacent Jafar analyses remain in the experiment archive; no new full profiling run was performed for this extraction.

Local validation passed: ./gradlew spotlessCheck checkstyleMain checkstyleTest, :microbench:compileJava, and 28 scoped test executions (12 scheduler, 2 consumer close/seek, 9 listener/ordering/Shared/partitioned, 4 Key_Shared ordering and 1 custom executor). No failures or skips; retries disabled with -PtestRetryCount=0, using --max-workers=2 --no-parallel.

• Make sure that the change passes the CI checks.

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model — coalesces internal executor submissions; retains serial FIFO draining and existing application callback executors.
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment