jerrypeng commented on code in PR #59523:
URL: https://github.com/apache/spark/pull/59523#discussion_r4408454632
##########
core/src/test/scala/org/apache/spark/shuffle/local/pipelined/PipelinedChannelShuffleSuite.scala:
##########
@@ -658,6 +659,52 @@ class PipelinedChannelShuffleSuite extends SparkFunSuite {
}
}
+ for (cachedPartitions <- Seq(0, 1, 2)) {
+ test(s"consumer cache is rejected before channel tasks start
(cached=$cachedPartitions)") {
+ withPipelinedSparkContext(cores = 4) { sc =>
+ val input = sc.parallelize(0 until 1000, 2).map(i => (i, i))
+ val consumer = new PipelinedShuffledRDD[Int, Int, Int](input, new
HashPartitioner(2))
+ .cache()
+ val blocks = (0 until cachedPartitions).map(RDDBlockId(consumer.id, _))
+ val blockManager = sc.env.blockManager
+ // Populate blocks without running the shuffle so a partial cache
cannot leave a
+ // producer waiting for a reader. Admission must reject before
starting either stage.
+ blocks.foreach { block =>
+ assert(blockManager.putIterator(
+ block, Iterator((block.splitIndex, block.splitIndex)),
+ StorageLevel.MEMORY_ONLY, tellMaster = true))
+ }
+
assert(blockManager.master.getLocations(blocks.toArray[BlockId]).forall(_.nonEmpty))
+
+ val error = intercept[SparkException](consumer.collect())
+ assert(error.getCondition === "PIPELINED_SHUFFLE_UNSUPPORTED")
+ assert(error.getMessage.contains("unpersist cached RDDs"))
+ assert(sc.statusTracker.getActiveStageIds().isEmpty)
Review Comment:
Agreed. The active-stage and queue snapshots could also be empty after a
late failure and cleanup. The test now registers JobEvents before collect(),
drains the listener bus after rejection, and asserts that both taskStarts and
stageEnds are empty. The exact error, queue-cleanup, and successful execution
after unpersist() checks remain.
I also corrected the setup comment to explain why cached blocks are
populated without executing the shuffle. The rejection comes from cache
preflight, before capacity admission; the four-task demand fits local[4].
Cache-location checks now cover every consumer partition and verify the
expected presence or absence of each block, so cached=0 explicitly checks that
both partitions have no cached blocks.
--
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]
