[SPARK-11617] [network] Fix leak in TransportFrameDecoder.

[vanzin]

The code was using the wrong API to add data to the internal composite
buffer, causing buffers to leak in certain situations. Use the right
API and enhance the tests to catch memory leaks.

Also, avoid reusing the composite buffers when downstream handlers keep
references to them; this seems to cause a few different issues even though
the ref counting code seems to be correct, so instead pay the cost of copying
a few bytes when that situation happens.

[JoshRosen]

Jenkins, retest this please.

[SparkQA]

Test build #45603 has finished for PR 9619 at commit ed0c1d7.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * public class JavaLBFGSExample\n * class LDA @Since(\"1.6.0\") (\n * case class Metadata(\n * require(className == expectedClassName, s\"Error loading metadata: Expected class name\" +\n * class SlidingRDDPartition[T](val idx: Int, val prev: Partition, val tail: Seq[T], val offset: Int)\n * class SlidingRDD[T: ClassTag](@transient val parent: RDD[T], val windowSize: Int, val step: Int)\n

[SparkQA]

Test build #45608 has finished for PR 9619 at commit ed0c1d7.

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

[SparkQA]

Test build #45779 has finished for PR 9619 at commit 8a7d194.

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

[vanzin]

/cc @zsxwing @rxin

[SparkQA]

Test build #45909 has finished for PR 9619 at commit 180456c.

• This patch fails from timeout after a configured wait of 250m.
• This patch merges cleanly.
• This patch adds no public classes.

[vanzin]

retest this please

[SparkQA]

Test build #45999 has finished for PR 9619 at commit 180456c.

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

[zsxwing]

@vanzin could you give a real case? Or this is just for correctness, even if downstream in Spark doesn't use retain?

[vanzin]

Actually Spark does use retain() when fetching shuffle blocks, and for some reason that causes problems. I think the real problem is somewhere in netty code, but this is the workaround the netty code itself uses (see ByteToMessageDecoder).

[zsxwing]

Yeah. Just saw retain() in ChunkFetchSuccess.

[zsxwing]

LGTM pending tests.

[SparkQA]

Test build #46015 has finished for PR 9619 at commit dcdfc31.

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

[zsxwing]

retest this please

[zsxwing]

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

This is a bug that will be fixed in #9707

[SparkQA]

Test build #46019 has finished for PR 9619 at commit 7fe9617.

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

[SparkQA]

Test build #46024 has finished for PR 9619 at commit 7fe9617.

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

[vanzin]

Merging this (master / 1.6).