mridulm commented on a change in pull request #33034:
URL: https://github.com/apache/spark/pull/33034#discussion_r656720067
##########
File path: core/src/main/scala/org/apache/spark/Dependency.scala
##########
@@ -122,6 +119,14 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C:
ClassTag](
*/
private[this] var _shuffleMergedFinalized: Boolean = false
+ /**
+ * shuffleSequenceId is used to give temporal ordering to the executions of
a ShuffleDependency.
+ * This is required in order to handle indeterministic stage retries for
push-based shuffle.
+ */
+ private[this] var nextShuffleSequenceId: Int = 0
+
+ def shuffleSequenceId: Int = nextShuffleSequenceId
Review comment:
This is not yet getting set ?
##########
File path: core/src/main/scala/org/apache/spark/Dependency.scala
##########
@@ -148,6 +153,18 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C:
ClassTag](
}
}
+ def resetShuffleMergeState(): Unit = {
+ _shuffleMergeEnabled = canShuffleMergeBeEnabled()
+ _shuffleMergedFinalized = false
+ mergerLocs = Nil
Review comment:
Why are we resetting `mergeLocs` ?
##########
File path: core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
##########
@@ -2057,7 +2058,7 @@ private[spark] class DAGScheduler(
// TODO: SPARK-35536: Cancel finalizeShuffleMerge if the stage is
cancelled
// TODO: during shuffleMergeFinalizeWaitSec
shuffleClient.finalizeShuffleMerge(shuffleServiceLoc.host,
- shuffleServiceLoc.port, shuffleId,
+ shuffleServiceLoc.port, shuffleId, stage.id,
Review comment:
Replace `stage.id` with the shuffle sequence id
--
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:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
