[SPARK-3453] Netty-based BlockTransferService, extracted from Spark core

[aarondav]

This PR encapsulates #2330, which is itself a continuation of #2240. The first goal of this PR is to provide an alternate, simpler implementation of the ConnectionManager which is based on Netty.

In addition to this goal, however, we want to resolve SPARK-3796, which calls for a standalone shuffle service which can be integrated into the YARN NodeManager, Standalone Worker, or on its own. This PR makes the first step in this direction by ensuring that the actual Netty service is as small as possible and extracted from Spark core. Given this, we should be able to construct this standalone jar which can be included in other JVMs without incurring significant dependency or runtime issues. The actual work to ensure that such a standalone shuffle service would work in Spark will be left for a future PR, however.

In order to minimize dependencies and allow for the service to be long-running (possibly much longer-running than Spark, and possibly having to support multiple version of Spark simultaneously), the entire service has been ported to Java, where we have full control over the binary compatibility of the components and do not depend on the Scala runtime or version.

These issues: have been addressed by folding in #2330:

SPARK-3453: Refactor Netty module to use BlockTransferService interface
SPARK-3018: Release all buffers upon task completion/failure
SPARK-3002: Create a connection pool and reuse clients across different threads
SPARK-3017: Integration tests and unit tests for connection failures
SPARK-3049: Make sure client doesn't block when server/connection has error(s)
SPARK-3502: SO_RCVBUF and SO_SNDBUF should be bootstrap childOption, not option
SPARK-3503: Disable thread local cache in PooledByteBufAllocator

TODO before mergeable:

• Implement uploadBlock()
• Unit tests for RPC side of code
• Performance testing (see comments here)
• Turn OFF by default (currently on for unit testing)

[SparkQA]

QA tests have started for PR 2753 at commit c0cd242.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2753 at commit c0cd242.

• This patch fails to build.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/21580/Test FAILed.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22415/
Test FAILed.

[aarondav]

Jenkins, retest this please.

[SparkQA]

Test build #22421 has started for PR 2753 at commit d7be11b.

• This patch merges cleanly.

[SparkQA]

Test build #22421 has finished for PR 2753 at commit d7be11b.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22421/
Test PASSed.

[aarondav]

That is a pass with netty turned on. Now I am turning it off for preparation to merge.

[SparkQA]

Test build #22427 has started for PR 2753 at commit cadfd28.

• This patch merges cleanly.

[SparkQA]

Test build #22427 has finished for PR 2753 at commit cadfd28.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22427/
Test FAILed.

[aarondav]

Jenkins, retest this please.

[SparkQA]

Test build #22437 has started for PR 2753 at commit cadfd28.

• This patch merges cleanly.

[SparkQA]

Test build #22437 has finished for PR 2753 at commit cadfd28.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/22437/
Test PASSed.

[rxin]

nit - it might be slightly more clear to write it this way

if (cachedClient != null) {
  if (cachedClient.isActive) {
    return cachedClient;
  } else {
    connectionPool.remove(address, cachedClient); // Remove inactive clients.
  }
}

You can do it in the next PR.

[rxin]

Merging this in master. Thanks!