viirya commented on code in PR #58097:
URL: https://github.com/apache/spark/pull/58097#discussion_r3817076228
##########
core/src/main/scala/org/apache/spark/ContextCleaner.scala:
##########
@@ -262,7 +262,23 @@ private[spark] class ContextCleaner(
listeners.asScala.foreach(_.shuffleCleaned(shuffleId))
logDebug("Cleaned pipelined shuffle " + shuffleId)
} else {
- logDebug("Asked to cleanup non-existent shuffle (maybe it was already
removed)")
+ // The shuffle is in NEITHER tracker. This is either a genuinely
non-existent shuffle
+ // (already removed) OR an in-process pipelined shuffle whose manager
keeps NO output
+ // tracker
(PipelinedChannelShuffleManager.usesStreamingShuffleOutputTracker = false):
+ // such a shuffle registers with no tracker at all, so the two
branches above miss it,
+ // yet its process-wide rendezvous queues (ChannelShuffleRendezvous)
still need freeing.
+ // Call shuffleDriverComponents.removeShuffle unconditionally: the
RemoveShuffle it issues
+ // routes to SparkEnv.unregisterShuffleFromAllManagers, which reaches
that manager's
+ // unregisterShuffle (-> ChannelShuffleRendezvous.removeShuffle) and
frees the queues.
+ // Safe for a truly non-existent id:
BlockManagerMasterEndpoint.removeShuffle finds no
+ // blocks and unregisterShuffle for an unknown id is a no-op on every
manager. No
+ // shuffle-manager type check here on purpose -- the cleaner stays
transport-agnostic and
+ // just balances the registerShuffle the ShuffleDependency constructor
issues for every
+ // shuffle.
+ logDebug("Cleaning tracker-less shuffle " + shuffleId)
+ shuffleDriverComponents.removeShuffle(shuffleId, blocking)
Review Comment:
Fixed, and thank you -- this contradicting "No behavior change by default"
was the right call. The else arm is now gated on
`!SparkEnv.get.pipelinedShuffleManager.usesStreamingShuffleOutputTracker` (a
tracker-less pipelined manager being active), so an already-cleaned regular
shuffle reaching it with the feature off is a no-op again -- no extra
cluster-wide `RemoveShuffle` RPC, no extra `shuffleCleaned` callback. Kept it a
capability check rather than a concrete-type check so the cleaner stays
transport-agnostic.
--
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]
