venkata91 commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3908866111
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1044,19 +1059,41 @@ private[spark] class MapOutputTrackerMaster(
/**
* Removes all shuffle outputs associated with this host. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists).
+ *
+ * When `skipReliablyStored` is true (executor/worker loss rather than a
fetch failure),
+ * shuffles whose output is reliably stored off-executor are left intact,
since losing the host
+ * does not lose their output.
*/
- def removeOutputsOnHost(host: String): Unit = {
- shuffleStatuses.valuesIterator.foreach { _.removeOutputsOnHost(host) }
+ def removeOutputsOnHost(host: String): Unit =
+ removeOutputsOnHost(host, skipReliablyStored = false)
+
+ def removeOutputsOnHost(host: String, skipReliablyStored: Boolean): Unit = {
+ shuffleStatuses.valuesIterator.foreach { status =>
+ if (!(skipReliablyStored && status.isReliablyStored)) {
+ status.removeOutputsOnHost(host)
+ }
+ }
incrementEpoch()
Review Comment:
Addressed it. Yes, we can avoid the `incrementEpoch` if all the shuffles are
reliably stored during an executor loss.
Thanks for your review. Long time back, I have worked on this layer for
push-based-shuffle. It is been a while now, looks like lot has changed. :)
--
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]
