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 7334ee03bd9a [SPARK-57711][SQL][TEST] Deflake
SQLAppStatusListenerMemoryLeakSuite "no memory leak"
7334ee03bd9a is described below
commit 7334ee03bd9ad59a65712aacdc09cf61c42d306d
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Sat Jun 27 11:08:02 2026 +0900
[SPARK-57711][SQL][TEST] Deflake SQLAppStatusListenerMemoryLeakSuite "no
memory leak"
### What changes were proposed in this pull request?
Wrap the final `noLiveData()` assertion in
`SQLAppStatusListenerMemoryLeakSuite`
("no memory leak") in an `eventually(...)` block so it polls for the
trailing
cleanup instead of asserting once immediately.
### Why are the changes needed?
The test is flaky on the SBT `sql - other tests` job, failing with
`noLiveData() was false`:
```
assert(statusStore.listener.get.noLiveData())
```
`SQLAppStatusListener` removes an execution's entries from `liveExecutions`
/
`stageMetrics` only once its end-event count reaches `jobs.size + 1`, i.e.
once
all of its `SparkListenerJobEnd` events and its
`SparkListenerSQLExecutionEnd`
have been processed.
The test runs 100 failing jobs (`df.foreach { throw ... }`). For a failed
job,
`DAGScheduler.failJobAndIndependentStages` notifies the job waiter -- which
unblocks the failing action on the test thread -- **before** it posts
`SparkListenerJobEnd` to the listener bus:
```scala
cleanupStateForJobAndIndependentStages(job)
job.listener.jobFailed(error) // unblocks
the action
listenerBus.post(SparkListenerJobEnd(job.jobId, ..., JobFailed(error))) //
posted afterwards
```
So the trailing `JobEnd` for the last failed execution can still be in
flight on
the DAGScheduler event-loop thread when the test thread races ahead, exits
the
loop, and calls `sc.listenerBus.waitUntilEmpty()`. If that `JobEnd` is
enqueued
just after the bus is drained, the failed execution never reaches the
cleanup
threshold and lingers in `liveExecutions`, so asserting `noLiveData()`
immediately intermittently observes leftover live data.
Note this is unrelated to the `kvstore.doAsync` metrics aggregation in
`onExecutionEnd`: the test sets `ASYNC_TRACKING_ENABLED=false`, so that
callback
runs inline on the listener thread (`sameThreadExecutorService`) and is
already
covered by `waitUntilEmpty()`. The residual race is purely in the delivery
of
the end events.
Polling with `eventually` lets the trailing event get delivered and the live
entries drain without weakening the assertion. The timeout is kept modest
(`5.seconds`) since the trailing event lands within milliseconds in
practice.
### Does this PR introduce _any_ user-facing change?
No, test only.
### How was this patch tested?
Existing test, run repeatedly. This reproduces under SBT, so the standard
fork
"Build" (SBT) exercises it.
### CI evidence
- Before (red): https://github.com/apache/spark/actions/runs/28000645341
(master SBT, `sql - other tests` -- `noLiveData() was false`)
- Also red: https://github.com/apache/spark/actions/runs/28004073554
- After (green):
https://github.com/HyukjinKwon/spark/actions/runs/28204756051
### Was this patch authored or co-authored using generative AI tooling?
Yes, drafted with assistance from Isaac.
Closes #56790 from HyukjinKwon/ci-fix/agent5-sqlappstatus-leak.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 4cfb2e8b16a7cae252512e7d43048f34bc0f3578)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../sql/execution/ui/SQLAppStatusListenerSuite.scala | 16 ++++++++++++++--
1 file changed, 14 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
index 3be048120496..6af000519aad 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
@@ -25,6 +25,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.json4s.jackson.JsonMethods._
import org.scalatest.BeforeAndAfter
+import org.scalatest.concurrent.Eventually._
import org.scalatest.time.SpanSugar._
import org.apache.spark._
@@ -1119,8 +1120,19 @@ class SQLAppStatusListenerMemoryLeakSuite extends
SparkFunSuite {
val statusStore = spark.sharedState.statusStore
assert(statusStore.executionsCount() <= 50)
assert(statusStore.planGraphCount() <= 50)
- // No live data should be left behind after all executions end.
- assert(statusStore.listener.get.noLiveData())
+ // No live data should be left behind after all executions end. A SQL
execution's live
+ // entries are removed only once its end-event count reaches jobs.size
+ 1 (the
+ // SparkListenerJobEnd(s) plus SparkListenerSQLExecutionEnd). For a
failed job the
+ // DAGScheduler notifies the job waiter -- unblocking the failing
action on this thread --
+ // *before* it posts SparkListenerJobEnd to the listener bus (see
+ // DAGScheduler.failJobAndIndependentStages). That trailing JobEnd can
therefore still be in
+ // flight when this thread calls waitUntilEmpty() above; if it is
enqueued just after the bus
+ // is drained, the failed execution never reaches the cleanup
threshold and lingers in
+ // liveExecutions. Poll with eventually() so the trailing end event is
delivered and the live
+ // entries drain, rather than asserting once immediately.
+ eventually(timeout(5.seconds), interval(10.milliseconds)) {
+ assert(statusStore.listener.get.noLiveData())
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
