[SPARK-28839][CORE] Avoids NPE in context cleaner when dynamic allocation and shuffle service are on

[HyukjinKwon]

What changes were proposed in this pull request?

This PR proposes to avoid to thrown NPE at context cleaner when shuffle service is on - it is kind of a small followup of #24817

Seems like it sets null for shuffleIds to track when the service is on. Later, removeShuffle tries to remove an element at shuffleIds which leads to NPE. It fixes it by explicitly not sending the event (ShuffleCleanedEvent) in this case.

See the code path below:

spark/core/src/main/scala/org/apache/spark/SparkContext.scala

Line 584 in cbad616

_cleaner.foreach(_.start())

spark/core/src/main/scala/org/apache/spark/ContextCleaner.scala

Line 125 in cbad616

cleaningThread.start()

spark/core/src/main/scala/org/apache/spark/ContextCleaner.scala

Line 190 in cbad616

doCleanupShuffle(shuffleId, blocking = blockOnShuffleCleanupTasks)

spark/core/src/main/scala/org/apache/spark/ContextCleaner.scala

Lines 220 to 230 in cbad616

	def doCleanupShuffle(shuffleId: Int, blocking: Boolean): Unit = {
	try {
	logDebug("Cleaning shuffle " + shuffleId)
	mapOutputTrackerMaster.unregisterShuffle(shuffleId)
	blockManagerMaster.removeShuffle(shuffleId, blocking)
	listeners.asScala.foreach(_.shuffleCleaned(shuffleId))
	logInfo("Cleaned shuffle " + shuffleId)
	} catch {
	case e: Exception => logError("Error cleaning shuffle " + shuffleId, e)
	}
	}

spark/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

Lines 353 to 357 in cbad616

	override def shuffleCleaned(shuffleId: Int): Unit = {
	// Because this is called in a completely separate thread, we post a custom event to the
	// listener bus so that the internal state is safely updated.
	listenerBus.post(ShuffleCleanedEvent(shuffleId))
	}

spark/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

Line 347 in cbad616

case ShuffleCleanedEvent(id) => cleanupShuffle(id)

spark/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

Lines 400 to 406 in cbad616

	private def cleanupShuffle(id: Int): Unit = {
	logDebug(s"Cleaning up state related to shuffle $id.")
	shuffleToActiveJobs -= id
	executors.asScala.foreach { case (_, exec) =>
	exec.removeShuffle(id)
	}
	}

spark/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

Line 475 in cbad616

if (shuffleIds.remove(id) && shuffleIds.isEmpty) {

spark/core/src/main/scala/org/apache/spark/scheduler/dynalloc/ExecutorMonitor.scala

Line 427 in cbad616

private val shuffleIds = if (shuffleTrackingEnabled) new mutable.HashSet[Int]() else null

Why are the changes needed?

This is a bug fix.

Does this PR introduce any user-facing change?

It prevents the exception:

19/08/21 06:44:01 ERROR AsyncEventQueue: Listener ExecutorMonitor threw an exception
java.lang.NullPointerException
	at org.apache.spark.scheduler.dynalloc.ExecutorMonitor$Tracker.removeShuffle(ExecutorMonitor.scala:479)
	at org.apache.spark.scheduler.dynalloc.ExecutorMonitor.$anonfun$cleanupShuffle$2(ExecutorMonitor.scala:408)
	at org.apache.spark.scheduler.dynalloc.ExecutorMonitor.$anonfun$cleanupShuffle$2$adapted(ExecutorMonitor.scala:407)
	at scala.collection.Iterator.foreach(Iterator.scala:941)
	at scala.collection.Iterator.foreach$(Iterator.scala:941)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1429)
	at scala.collection.IterableLike.foreach(IterableLike.scala:74)
	at scala.collection.IterableLike.foreach$(IterableLike.scala:73)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:56)
	at org.apache.spark.scheduler.dynalloc.ExecutorMonitor.cleanupShuffle(ExecutorMonitor.scala:407)
	at org.apache.spark.scheduler.dynalloc.ExecutorMonitor.onOtherEvent(ExecutorMonitor.sc

How was this patch test?

Unittest was added.

[HyukjinKwon]

@vanzin do you mind if I ask to take a look and see if it makes sense?

[dongjoon-hyun]

So, is this only for 3.0?

[HyukjinKwon]

Yup, I think so.

[SparkQA]

Test build #109568 has finished for PR 25551 at commit 4363ef1.

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

[HyukjinKwon]

retest this please

[SparkQA]

Test build #109570 has finished for PR 25551 at commit 4363ef1.

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

[HyukjinKwon]

retest this please

[SparkQA]

Test build #109577 has finished for PR 25551 at commit 4363ef1.

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

[viirya]

good catch!

[SparkQA]

Test build #109599 has finished for PR 25551 at commit 4f5de95.

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

[SparkQA]

Test build #109637 has finished for PR 25551 at commit c8b0da9.

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

[SparkQA]

Test build #4839 has finished for PR 25551 at commit c8b0da9.

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

[wangyum]

retest this please

[SparkQA]

Test build #109646 has finished for PR 25551 at commit c8b0da9.

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

[vanzin]

Merging to master.

[vanzin]

You could have used mockito's verify instead of this, but this is ok.

[HyukjinKwon]

Thanks all!