[SPARK-9193] Avoid assigning tasks to "lost" executor(s)

[GraceH]

Now, when some executors are killed by dynamic-allocation, it leads to some mis-assignment onto lost executors sometimes. Such kind of mis-assignment causes task failure(s) or even job failure if it repeats that errors for 4 times.

The root cause is that _killExecutors_ doesn't remove those executors under killing ASAP. It depends on the _OnDisassociated_ event to refresh the active working list later. The delay time really depends on your cluster status (from several milliseconds to sub-minute). When new tasks to be scheduled during that period of time, it will be assigned to those "active" but "under killing" executors. Then the tasks will be failed due to "executor lost". The better way is to exclude those executors under killing in the makeOffers(). Then all those tasks won't be allocated onto those executors "to be lost" any more.

[SparkQA]

Test build #37814 has finished for PR 7528 at commit b5546ce.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class IllegalArgumentException(Exception):
  • case class DayOfYear(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
  • case class DayOfMonth(child: Expression) extends UnaryExpression with ImplicitCastInputTypes
  • case class ConcatWs(children: Seq[Expression])

[squito]

Hi @GraceH thanks for finding this bug and posting the PR. The changes look completely reasonable, but would it be possible to add a unit test? I know this is a big ask, since it doesn't look like there are any tests right now on that test anything like this for CoarseGrainedSchedulerBackend, so it will probably take some work to do.

I believe this changes are correct, so the test isn't so much about verifying this fix as it is to prevent future regressions. It would also really help in the future for following along the code -- eg., it took me a while to figure out how anything was ever removed from executorsPendingToRemove.

[GraceH]

@squito Thanks for the confirmation. If you do think it is necessary to add unit test for CoarseGrainedSchedulerBackend, one more JIRA can be filed for that. Because it seems that should cover more than what this PR does. What do you think?

[andrewor14]

This is a little hard to read. Could you rewrite this as follows:

private def makeOffers() {
  val activeExecutors = executorDataMap.filterKeys(!executorsPendingToRemove)
  val workerOffers = activeExecutors.map { case (id, executorData) =>
    new WorkerOffer(id, ...)
  }
  launchTasks(scheduler.resourceOffers(workerOffers))
}

[GraceH]

Good point. Will re-organize the code snippet.

[andrewor14]

Hi @GraceH this looks great. I left a few minor wording suggestions to improve readability.

As for unit tests, I do agree that we should have them, but it seems outside the scope of this fix since properly testing all of this logic would require a non-trivial refactor. In this case, I am more inclined to merge in this fix than to delay it indefinitely.

[GraceH]

Thanks @andrewor14 . That makes sense. Will have a revised version with more readable lines.

[SparkQA]

Test build #37953 has finished for PR 7528 at commit 6e2ed96.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[squito]

yeah, I don't love the idea of adding things w/out tests, but in this case I suppose its best left for the future. lgtm pending the tests passing

[GraceH]

Thanks @squito.

[SparkQA]

Test build #37954 has finished for PR 7528 at commit ecc1da6.

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