ivoson commented on code in PR #54136:
URL: https://github.com/apache/spark/pull/54136#discussion_r2785944712
##########
core/src/main/scala/org/apache/spark/storage/BlockManager.scala:
##########
@@ -773,6 +802,10 @@ private[spark] class BlockManager(
if (blockId.isShuffle) {
logDebug(s"Putting shuffle block ${blockId}")
+ // Wait for ShuffleManager to be initialized before handling shuffle
migration requests.
+ // This can happen when an executor receives migration requests before
its ShuffleManager
+ // is fully initialized.
+ waitForShuffleManagerInit(blockId)
Review Comment:
We need to register blockManager then start heartbeat thread, waiting
shuffle manager to be initialized will need to wait for updateDependencies. It
will take long time, and there could be heartbeat timeout.
I moved the `waitForShuffleManagerInit` in the `lazy shuffleManager`
initialization block for two reasons:
1. Avoid the repeated check after the lazy shuffleManager get successfully
initialized;
2. Could cover other future potential cases to initialize shuffleManager as
NULL;
This way we can avoid adding new RPCs or introduce more complexity in driver
side.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
