[SPARK-4342] [Core]connection ack timeout improvement, replace Timer with ScheudledExecutor...

[haitaoyao]

• replace java.util.Timer with ScheduledExectorService
• avoid reference to the whole message, use message id only.

[AmplabJenkins]

Can one of the admins verify this patch?

[zsxwing]

Is there any reason for the magic number 2?

[haitaoyao]

to avoid some task block the whole scheduler. ScheduledExecutorService executes the task in its own thread pool, instead of fork a new thread for the task execution.

[zsxwing]

I just checked the codes:

    public ScheduledThreadPoolExecutor(int corePoolSize,
                                       ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

2 is the min thread number, not the max thread number.

[zsxwing]

corePoolSize the number of threads to keep in the pool, even if they are idle

Here I think 1 is enough.

[JoshRosen]

This looks like a duplicate of another JIRA, which was resolved by #3259, so do you mind closing this? Thanks!

(Sorry that I missed this PR; I guess it got buried in a backlog of other stuff)