zhenlineo commented on code in PR #42009:
URL: https://github.com/apache/spark/pull/42009#discussion_r1280117379
##########
connector/connect/client/jvm/src/test/scala/org/apache/spark/sql/SparkSessionE2ESuite.scala:
##########
@@ -96,5 +103,148 @@ class SparkSessionE2ESuite extends RemoteSparkSession {
assert(e2.getMessage.contains("OPERATION_CANCELED"), s"Unexpected
exception: $e2")
finished = true
assert(ThreadUtils.awaitResult(interruptor, 10.seconds))
+ assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
+ }
+
+ test("interrupt tag") {
+ val session = spark
+ import session.implicits._
+
+ // global ExecutionContext has only 2 threads in Apache Spark CI
+ // create own thread pool for four Futures used in this test
+ val numThreads = 4
+ val fpool = ThreadUtils.newForkJoinPool("job-tags-test-thread-pool",
numThreads)
+ val executionContext = ExecutionContext.fromExecutorService(fpool)
+
+ val q1 = Future {
+ assert(spark.getTags() == Set())
+ spark.addTag("two")
+ assert(spark.getTags() == Set("two"))
+ spark.clearTags() // check that clearing all tags works
+ assert(spark.getTags() == Set())
+ spark.addTag("one")
+ assert(spark.getTags() == Set("one"))
+ try {
+ spark
+ .range(10)
+ .map(n => {
+ Thread.sleep(30000); n
+ })
+ .collect()
+ } finally {
+ spark.clearTags() // clear for the case of thread reuse by another
Future
+ }
+ }(executionContext)
+ val q2 = Future {
+ assert(spark.getTags() == Set())
+ spark.addTag("one")
+ spark.addTag("two")
+ spark.addTag("one")
+ spark.addTag("two") // duplicates shouldn't matter
+ assert(spark.getTags() == Set("one", "two"))
+ try {
+ spark
+ .range(10)
+ .map(n => {
+ Thread.sleep(30000); n
+ })
+ .collect()
+ } finally {
+ spark.clearTags() // clear for the case of thread reuse by another
Future
+ }
+ }(executionContext)
+ val q3 = Future {
+ assert(spark.getTags() == Set())
+ spark.addTag("foo")
+ spark.removeTag("foo")
+ assert(spark.getTags() == Set()) // check that remove works removing the
last tag
+ spark.addTag("two")
+ assert(spark.getTags() == Set("two"))
+ try {
+ spark
+ .range(10)
+ .map(n => {
+ Thread.sleep(30000); n
+ })
+ .collect()
+ } finally {
+ spark.clearTags() // clear for the case of thread reuse by another
Future
+ }
+ }(executionContext)
+ val q4 = Future {
+ assert(spark.getTags() == Set())
+ spark.addTag("one")
+ spark.addTag("two")
+ spark.addTag("two")
+ assert(spark.getTags() == Set("one", "two"))
+ spark.removeTag("two") // check that remove works, despite duplicate add
+ assert(spark.getTags() == Set("one"))
+ try {
+ spark
+ .range(10)
+ .map(n => {
+ Thread.sleep(30000); n
+ })
+ .collect()
+ } finally {
+ spark.clearTags() // clear for the case of thread reuse by another
Future
+ }
+ }(executionContext)
+ val interrupted = mutable.ListBuffer[String]()
+
+ // q2 and q3 should be cancelled
+ interrupted.clear()
+ eventually(timeout(20.seconds), interval(1.seconds)) {
+ val ids = spark.interruptTag("two")
+ interrupted ++= ids
+ assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
+ }
+ val e2 = intercept[SparkException] {
+ ThreadUtils.awaitResult(q2, 1.minute)
+ }
+ assert(e2.getCause.getMessage contains "OPERATION_CANCELED")
+ val e3 = intercept[SparkException] {
+ ThreadUtils.awaitResult(q3, 1.minute)
+ }
+ assert(e3.getCause.getMessage contains "OPERATION_CANCELED")
+ assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
+
+ // q1 and q4 should be cancelled
+ interrupted.clear()
+ eventually(timeout(20.seconds), interval(1.seconds)) {
+ val ids = spark.interruptTag("one")
+ interrupted ++= ids
+ assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
+ }
+ val e1 = intercept[SparkException] {
+ ThreadUtils.awaitResult(q1, 1.minute)
+ }
+ assert(e1.getCause.getMessage contains "OPERATION_CANCELED")
+ val e4 = intercept[SparkException] {
+ ThreadUtils.awaitResult(q4, 1.minute)
+ }
+ assert(e4.getCause.getMessage contains "OPERATION_CANCELED")
+ assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
+ }
+
+ test("interrupt operation") {
Review Comment:
@LuciferYang Our current solution only stubs classes from user session
scoped libs. That covers jars added via `session.addArtifact`. The bug
Spark-44576 should be fixed as streaming should not fail when user adds session
scoped libs. If the failing tests really bothers you, we can also move them
into separate test file for now. The cause is the unnecessary dependency pulled
by lambdas inside the same class file.
--
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]
