[SPARK-4195][Core]retry to fetch blocks's result when fetchfailed's reason is connection timeout

[lianhuiwang]

when there are many executors in a application(example:1000),Connection timeout often occure.
Exception is:
WARN nio.SendingConnection: Error finishing connection
java.net.ConnectException: Connection timed out
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:739)
at org.apache.spark.network.nio.SendingConnection.finishConnect(Connection.scala:342)
at org.apache.spark.network.nio.ConnectionManager$$anon$11.run(ConnectionManager.scala:273)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)

that will make driver as these executors are lost, but in fact these executors are alive. so add retry mechanism to reduce the probability of the occurrence of this problem. @rxin

[SparkQA]

Test build #22761 has started for PR 3061 at commit dcfef7d.

• This patch merges cleanly.

[SparkQA]

Test build #22761 has finished for PR 3061 at commit dcfef7d.

• 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/22761/
Test FAILed.

[rxin]

@aarondav we should incorporate this into the new transport thing

[JoshRosen]

To avoid memory leaks, we need to be sure that this won't grow without bound. Let me try to walk through the cases...

• If no errors occur, this will remain empty since entries are only added on error.
• If a fetch fails and a retry succeeds, then the entry is removed from this map.
• If the maximum number of attempts is exceeded, we don't remove an entry from this map.

So, looks like this adds a memory leak?

[lianhuiwang]

yes, thank you for finding its error.

[JoshRosen]

There are a bunch of minor style issues here; I don't want to comment on them individually, so please take a look at https://cwiki.apache.org/confluence/display/SPARK/Spark+Code+Style+Guide.

[JoshRosen]

Why not catch ConnectException? I suppose it's always safe to retry as long as the number of retries is bounded, but it's probably better to catch the narrower exception if we're only trying to deal with connection establishment errors.

[lianhuiwang]

because in https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/network/nio/ConnectionManager.scala#L963 doesnot catch ConnectException and throw a IOException. so now in here we only catch IOException. if we are only trying to deal with connection errors, it need to catch ConnectException in ConnectionManager.scala.

[SparkQA]

Test build #24757 has started for PR 3061 at commit 5a73b04.

• This patch merges cleanly.

[SparkQA]

Test build #24757 has finished for PR 3061 at commit 5a73b04.

• 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/24757/
Test FAILed.

[rxin]

Actually let's close this one. Based on Tencent's feedback, we have already implemented the same functionality in Netty. I'm not sure whether it is worth it to fix the current connection manager that very few people understand.

[lianhuiwang]

OK. I will close this PR.