jerrypeng commented on code in PR #59524:
URL: https://github.com/apache/spark/pull/59524#discussion_r4291190855
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -9534,6 +9738,27 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
}
}
+ private def assertRepeatedPipelinedConsumptionRejected(
+ result: RDD[_], shuffleIds: Seq[Int]): Unit = {
+ for (_ <- 0 until 2) {
+ clearInvocations(scheduler, mapOutputTracker)
+ val failure = submitAndCaptureFailure(result,
result.partitions.indices.toArray)
+ assertPipelinedUnsupported(failure, "more than one consumer")
Review Comment:
Thanks, this is valid. I tightened the three fan-out assertions to require
`(fan-out / branching)` and added the disjoint-partition consumer case. The
repeated-readhelper now checks the full structured reason, including the
expected shuffle ID and reducer partition. The multi-stage case pins the
upstream shuffle, and the aliasing case
pins reducer 0 rather than logical reader partition 1.
I also verified the regression signal with mutations: disabling fan-out
rejection fails all four fan-out tests, and changing the reported shuffle ID or
using the reader partition fails the corresponding scheduler test. All 115
pipelined scheduler tests and 26 channel shuffle tests pass; scalastyle is
clean. This validation freshly compiled the current scheduler, TaskSetManager,
and test sources against pinned prebuilt dependency jars.
--
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]
