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 test now registers JobEvents before collect(), drains the
listener bus after rejection, and asserts empty task-start and stage-completion
histories. The exact error, queue-cleanup, and recovery checks remain. I
corrected the setup comment to describe cache population without executing the
shuffle and now verify cache locations for every consumer partition, including
explicit absence in the cold-cache case.
--
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]
