Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20449#discussion_r166877282
--- Diff: core/src/test/scala/org/apache/spark/JobCancellationSuite.scala
---
@@ -320,6 +321,41 @@ class JobCancellationSuite extends SparkFunSuite with
Matchers with BeforeAndAft
f2.get()
}
+ test("Interruptible iterator of shuffle reader") {
+ import JobCancellationSuite._
+ sc = new SparkContext("local[2]", "test")
+
+ val f = sc.parallelize(1 to 1000, 2).map { i => (i, i) }
+ .repartitionAndSortWithinPartitions(new HashPartitioner(2))
+ .mapPartitions { iter =>
+ taskStartedSemaphore.release()
+ // Small delay to ensure that foreach is cancelled if task is
killed
+ Thread.sleep(1000)
--- End diff --
+1
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
