Fix read thread blocking in sendResponseAndWait causing READ_ENTRY_REQUEST p99 latency spike

[hangc0276]

Motivation

During performance testing, we observed that bookkeeper_server_READ_ENTRY p99 latency is less than 100ms, but bookkeeper_server_READ_ENTRY_REQUEST p99 latency spikes to 3 seconds. The two metrics
differ in scope:

• READ_ENTRY measures the storage read time only (from before readData() to after it completes)
• READ_ENTRY_REQUEST measures from request enqueue time (enqueueNanos) through response write

The root cause is in PacketProcessorBase.sendResponseAndWait(). When readWorkerThreadsThrottlingEnabled=true (the default), read responses go through sendResponseAndWait, which calls future.get() to
block the read thread pool thread until the Netty channel write completes.

Under heavy read load, this causes a cascading failure:

1. future.get() blocks each read thread waiting for Netty write completion
2. If writes are slow (client backpressure, network congestion, event loop busy), threads pile up blocked
3. The read thread pool becomes saturated — all threads blocked waiting for writes
4. New read requests queue up, and their scheduling delay grows
5. READ_ENTRY_REQUEST (measured from enqueueNanos) spikes, even for requests to healthy channels

One slow consumer can block a thread pool thread, reducing capacity for all other channels — a classic head-of-line blocking problem.

Changes

1. Non-blocking sendResponseAndWait (PacketProcessorBase.java)

Replace the blocking future.get() in sendResponseAndWait() with a non-blocking ChannelFutureListener. The key design decisions:

• Read thread freed immediately: After writeAndFlush, the thread returns to the pool and can process other requests instead of blocking on future.get()
• Throttling semantics preserved: onReadRequestFinish() (which releases the read semaphore) is moved into the ChannelFutureListener, so it is only called after the write completes. This ensures the
  read concurrency limit still gates on write completion, without wasting a thread to enforce it.
• Captured local variables: Since the processor may be recycled after sendResponseAndWait returns, enqueueNanos and requestProcessor are captured as local variables before the async callback.
• Semaphore always released: Both the success and failure paths in the listener call onReadRequestFinish(), preventing semaphore leaks on write failure. (The previous code had a subtle issue: on
  ExecutionException/InterruptedException, sendResponseAndWait returned early without recording the metric, though onReadRequestFinish was still called by the caller.)

2. Enable maxReadsInProgressLimit by default (ServerConfiguration.java)

Changed the default of maxReadsInProgressLimit from 0 (unlimited) to 10000.

With the old blocking sendResponseAndWait, the read thread pool (default 8 threads) acted as an implicit concurrency bound — each thread blocked on one write at a time, so at most 8 responses could be
buffered. With the new non-blocking approach, threads are freed immediately and keep processing reads. Without a bound, a slow consumer could cause unbounded response buffer growth in Netty's write
queue, leading to OOM.

The maxReadsInProgressLimit semaphore now serves as the explicit bound. Since onReadRequestFinish() is called in the ChannelFutureListener (after the write completes), the semaphore precisely counts
reads whose response data is in memory but not yet flushed. The default of 10,000 provides good throughput (~40MB at 4KB entries) while preventing unbounded memory growth. Users with larger entries
should tune this lower (memoryBudget / avgEntrySize).

Test Plan

Added 5 new unit tests to ReadEntryProcessorTest:

• testThrottledReadNonBlockingOnSuccess — verifies that with throttle=true, run() returns immediately and onReadRequestFinish() is deferred until the write future completes
• testThrottledReadNonBlockingOnWriteFailure — verifies onReadRequestFinish() is still called when the write fails, ensuring the read semaphore is always released
• testNonThrottledReadCallsOnFinishSynchronously — verifies that with throttle=false, onReadRequestFinish() is called synchronously
• testDefaultMaxReadsInProgressLimitIsEnabled — verifies maxReadsInProgressLimit defaults to 10000
• testThrottledReadHoldsSemaphoreUntilWriteCompletes — uses a real Semaphore(1) wired to onReadRequestStart/onReadRequestFinish to verify the full lifecycle: semaphore acquired at request creation,
  held through non-blocking run() return, blocks a second read while write is in progress, and released only when the write future completes

[wenbingshen]

LGTM. The change to maxReadsInProgressLimit needs to be added to the release documentation in the future.