RATIS-2629. Handle Netty based request asynchronously and improve exception handling

[spacemonkd]

What changes were proposed in this pull request?

RATIS-2629. Handle Netty based request asynchronously and improve exception handling

Currently NettyRpcService handles every inbound request synchronously on the Netty IO event-loop thread and its inbound handler does not override exceptionCaught.

This can cause:

• a slow or long-running request occupies the worker thread delaying all other requests on that event loop
• only IOException is converted into an error reply. Any other Throwable escapes the handler with no exceptionCaught, Netty's default handling only logs it and never writes a reply, so the client blocks until its request timeout.

This patch offloads the request handling out of the core IO path, and closes the channel to prevent blocking.
This mirrors how gRPC handles the requests.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2629

How was this patch tested?

Patch was tested via unit tests.

[spacemonkd]

@szetszwo could you take a look?

[ss77892]

So, it's still a single worker pool (corePoolSize is 0). But instead of using TCP backpressure, we are pushing everything to the unlimited queue and creating pressure on the memory. Moreover, previously, Netty’s worker EventLoops handled connection shards independently, so a blocked request delayed only that shard’s RPCs. The new default single request worker queues all inbound RPCs, including heartbeats, so one slow request can delay heartbeats and trigger leader election.

[spacemonkd]

Yes this was one issue which I missed and faced before (hence the test failure in flaky test suite).
corePoolSize=0 + an unbounded queue causes requestExecutor to be single-threaded, and since handle() blocks until commit, this causes the timeouts.
I have addressed this by switching to a fixed pool for now as the related change would increase LoC.

Filed https://issues.apache.org/jira/browse/RATIS-2637 for the improvement as a follow up.

[ss77892]

Backpressure control should be a part of these changes. You have removed the one that was going through TCP flow control and don't provide any replacement. Another thing: we have ThreadPoolExecutor with corePoolSize=0 and unbounded queue. Let me quote "Java Concurrency in Practice":
[3] Developers are sometimes tempted to set the core size to zero so that the worker threads will eventually be torn down and therefore won't prevent the JVM from exiting, but this can cause some strange‐seeming behavior in thread pools that don't use a SynchronousQueue for their work queue (as newCachedThreadPool does). If the pool is already at the core size, ThreadPoolExecutor creates a new thread only if the work queue is full. So tasks submitted to a thread pool with a work queue that has any capacity and a core size of zero will not execute until the queue fills up, which is usually not what is desired.
So, it's still a single thread.

[spacemonkd]

Another thing: we have ThreadPoolExecutor with corePoolSize=0 and unbounded queue.
the corePoolSize=0 + unbounded-queue cached pool is effectively single-threaded, however this PR now defaults to a fixed pool (corePoolSize=32), so inbound RPCs including heartbeats are no longer serialized behind one slow request.

So I have effectively switched back to the earlier behaviour for the time being.
Let's avoid making this patch bigger and address separately as right now this is effective the previous behaviour.