[SPARK-4020] Do not rely on timeouts to remove failed block managers

[andrewor14]

If an executor fails without being scheduled to run any tasks, then DAGScheduler won't notify BlockManagerMasterActor that the associated block manager should be removed. Instead, the associated block manager will be expired only after a few rounds of heartbeat timeouts. In terms of removal treatment, there should really be no distinction between executors that have been scheduled tasks and those that have not.

The fix, then, is to add all known executors to TaskSchedulerImpl's activeExecutorIds whether or not it has been scheduled a task. In fact, the existing comment above activeExecutorIds is

// Which executor IDs we have executors on
val activeExecutorIds = new HashSet[String]

not "Which executors have been scheduled tasks thus far."

[andrewor14]

Hey @kayousterhout can you take a look at this at your earliest convenience? This is blocking #2840.

[SparkQA]

QA tests have started for PR 2865 at commit ff3172b.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2865 at commit ff3172b.

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

[shaneknapp]

jenkins, test this please

[SparkQA]

QA tests have started for PR 2865 at commit ff3172b.

• This patch merges cleanly.

[SparkQA]

QA tests have finished for PR 2865 at commit ff3172b.

• 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/21941/
Test PASSed.

[kayousterhout]

@andrewor14 this looks good, and definitely seems to represent the "expected" use of activeExecutorIds.