This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 1306da8274f0 [SPARK-57656][CONNECT][TEST] Stabilize flaky
SparkSessionE2ESuite interrupt-all tests
1306da8274f0 is described below
commit 1306da8274f0add5e8b80ffa4517247873b6b52d
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Wed Jun 24 19:44:35 2026 +0900
[SPARK-57656][CONNECT][TEST] Stabilize flaky SparkSessionE2ESuite
interrupt-all tests
### What changes were proposed in this pull request?
Make the two `SparkSessionE2ESuite` "interrupt all" tests robust against
two flakiness sources:
1. **Class-fetch race.** Run each long-running typed `map` query through a
single call site and warm
it up once (`sleep=0`) before any interrupt, so the closure/`TypeTag`
artifact classes are already
loaded on the executor. Otherwise an `interruptAll()` landing during the
executor's first-time
remote fetch of `$$typecreatorNN` surfaces as `RemoteClassLoaderError`
instead of
`OPERATION_CANCELED`.
2. **Leaked interruptor / cascade.** Wrap the foreground-interrupt test
body in
`try/finally { finished = true }`. Previously, if an assertion failed,
the background `interruptor`
Future kept calling `interruptAll()` and canceled the operations of
*subsequent* tests — turning
one failure into a cascade of `OPERATION_CANCELED` failures across the
whole suite.
### Why are the changes needed?
`SparkSessionE2ESuite` intermittently fails in CI. Confirmed flaky (the
same module group passes on
other runs of the same commit).
**Before (failing in apache/spark CI):** one `RemoteClassLoaderError`
cascading into 7 failures —
https://github.com/apache/spark/actions/runs/28037082466/job/82997066297
**After (this change, validated on a fork):** full connect module `Total
1614, Failed 0`, and
`SparkSessionE2ESuite` re-run 8x with 0 failures —
https://github.com/HyukjinKwon/spark/actions/runs/28066097371 and
https://github.com/HyukjinKwon/spark/actions/runs/28074772169
### Does this PR introduce any user-facing change?
No, test-only.
### How was this patch tested?
Re-ran `SparkSessionE2ESuite` 8x plus the full connect module on CI (links
above); all green.
### Was this patch authored or co-authored using generative AI tooling?
Yes, drafted with Claude Code.
Closes #56728 from HyukjinKwon/SPARK-57656.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit dae1847a79d1c9929b40ff8945b951c2476d1d85)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/sql/connect/SparkSessionE2ESuite.scala | 46 +++++++++++++++++-----
1 file changed, 36 insertions(+), 10 deletions(-)
diff --git
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionE2ESuite.scala
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionE2ESuite.scala
index 184868a0df23..9c149a858018 100644
---
a/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionE2ESuite.scala
+++
b/sql/connect/client/jvm/src/test/scala/org/apache/spark/sql/connect/SparkSessionE2ESuite.scala
@@ -41,11 +41,20 @@ class SparkSessionE2ESuite extends ConnectFunSuite with
RemoteSparkSession {
val session = spark
import session.implicits._
implicit val ec: ExecutionContextExecutor = ExecutionContext.global
+ // Run the long-running query through a single call site, and warm it up
once before any
+ // interrupt. Otherwise the very first execution has to ship/fetch the map
closure and its
+ // TypeTag artifact classes to the executor on demand; if an interrupt
lands during that
+ // first-time remote class fetch, it surfaces as a RemoteClassLoaderError
instead of
+ // OPERATION_CANCELED, making this test flaky (see
SparkSessionE2ESuite$$typecreatorNN).
+ def runMapQuery(sleepMs: Long): Unit = {
+ spark.range(10).map(n => { Thread.sleep(sleepMs); n }).collect()
+ }
+ runMapQuery(0)
val q1 = Future {
- spark.range(10).map(n => { Thread.sleep(30000); n }).collect()
+ runMapQuery(30000)
}
val q2 = Future {
- spark.range(10).map(n => { Thread.sleep(30000); n }).collect()
+ runMapQuery(30000)
}
var q1Interrupted = false
var q2Interrupted = false
@@ -88,6 +97,16 @@ class SparkSessionE2ESuite extends ConnectFunSuite with
RemoteSparkSession {
@volatile var finished = false
val interrupted = mutable.ListBuffer[String]()
+ // Run the long-running query through a single call site, and warm it up
once before the
+ // background interruptor starts. Otherwise the first execution has to
ship/fetch the map
+ // closure and its TypeTag artifact classes to the executor on demand; if
an interrupt lands
+ // during that first-time remote class fetch, it surfaces as a
RemoteClassLoaderError instead
+ // of OPERATION_CANCELED, making this test flaky (see
SparkSessionE2ESuite$$typecreatorNN).
+ def runMapQuery(sleepMs: Long): Unit = {
+ spark.range(10).map(n => { Thread.sleep(sleepMs); n }).collect()
+ }
+ runMapQuery(0)
+
val interruptor = Future {
eventually(timeout(20.seconds), interval(1.seconds)) {
val ids = spark.interruptAll()
@@ -96,15 +115,22 @@ class SparkSessionE2ESuite extends ConnectFunSuite with
RemoteSparkSession {
}
finished
}
- val e1 = intercept[SparkException] {
- spark.range(10).map(n => { Thread.sleep(30.seconds.toMillis); n
}).collect()
- }
- assert(e1.getMessage.contains("OPERATION_CANCELED"), s"Unexpected
exception: $e1")
- val e2 = intercept[SparkException] {
- spark.range(10).map(n => { Thread.sleep(30.seconds.toMillis); n
}).collect()
+ try {
+ val e1 = intercept[SparkException] {
+ runMapQuery(30.seconds.toMillis)
+ }
+ assert(e1.getMessage.contains("OPERATION_CANCELED"), s"Unexpected
exception: $e1")
+ val e2 = intercept[SparkException] {
+ runMapQuery(30.seconds.toMillis)
+ }
+ assert(e2.getMessage.contains("OPERATION_CANCELED"), s"Unexpected
exception: $e2")
+ } finally {
+ // Always release the background interruptor. If an assertion above
fails, this prevents the
+ // interruptor Future from continuing to call interruptAll() and
canceling operations of the
+ // subsequent tests in this suite (which previously caused cascading
OPERATION_CANCELED
+ // failures across the whole suite).
+ finished = true
}
- assert(e2.getMessage.contains("OPERATION_CANCELED"), s"Unexpected
exception: $e2")
- finished = true
assert(awaitResult(interruptor, 10.seconds))
assert(interrupted.length == 2, s"Interrupted operations: $interrupted.")
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
