dongjoon-hyun commented on a change in pull request #28746:
URL: https://github.com/apache/spark/pull/28746#discussion_r440554399
##########
File path: core/src/main/scala/org/apache/spark/deploy/LocalSparkCluster.scala
##########
@@ -63,23 +65,34 @@ class LocalSparkCluster(
/* Start the Workers */
for (workerNum <- 1 to numWorkers) {
- val workerEnv = Worker.startRpcEnvAndEndpoint(localHostname, 0, 0,
coresPerWorker,
- memoryPerWorker, masters, null, Some(workerNum), _conf,
- conf.get(config.Worker.SPARK_WORKER_RESOURCE_FILE))
+ val (workerEnv, workerRef) =
Worker.startRpcEnvAndEndpoint(localHostname, 0, 0,
+ coresPerWorker, memoryPerWorker, masters, null, Some(workerNum), _conf,
+ conf.get(config.Worker.SPARK_WORKER_RESOURCE_FILE), isLocalCluster =
true)
workerRpcEnvs += workerEnv
+ workerRefs += workerRef
}
masters
}
def stop(): Unit = {
logInfo("Shutting down local Spark cluster.")
+ // SPARK-31922: make sure all the workers have handled the
messages(`KillExecutor`,
+ // `ApplicationFinished`) from the Master before we shutdown the workers'
rpcEnvs.
+ // Otherwise, we could hit "RpcEnv already stopped" error.
+ var busyWorkers = workerRefs
+ while (busyWorkers.nonEmpty) {
+ Thread.sleep(300)
+ busyWorkers =
busyWorkers.filterNot(_.askSync[Boolean](IsWorkerReadyToStop))
Review comment:
Thank you for working on this, @Ngone51 . BTW, this seems to cause an
infinite loop if a worker cannot response for some reasons. Can we use a finite
`for` statement instead of `while` statement to avoid any potential issues?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
