From e75ea82034215868c9cc327e83aab3a208d5a1f0 Mon Sep 17 00:00:00 2001 From: Hyukjin Kwon Date: Fri, 26 Jun 2026 07:32:37 +0900 Subject: [PATCH 1/2] [DO-NOT-MERGE][SQL][TEST] Deflake SQLAppStatusListenerMemoryLeakSuite "no memory leak" ### What changes were proposed in this pull request? Wrap the final `noLiveData()` assertion in `SQLAppStatusListenerMemoryLeakSuite` in an `eventually(...)` block. ### Why are the changes needed? The test fails intermittently on the SBT `sql - other tests` job (e.g. master runs 28000645341, 28004073554) with `noLiveData() was false`. The cleanup that removes entries from `liveExecutions`/`stageMetrics` is finalized by the metrics aggregation triggered on `SparkListenerSQLExecutionEnd`; asserting immediately after `waitUntilEmpty()` is racy. Polling with `eventually` removes the race without weakening the assertion. ### Does this PR introduce any user-facing change? No, test only. ### How was this patch tested? Existing test, repeated. Co-authored-by: Isaac --- .../sql/execution/ui/SQLAppStatusListenerSuite.scala | 10 ++++++++-- 1 file changed, 8 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 3be0481204967..179ba251754e6 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,13 @@ 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. The cleanup of live + // executions/stage metrics is finalized when the metrics aggregation triggered by the + // SQLExecutionEnd event completes, so wait for the listener to drain rather than asserting + // immediately to avoid a timing race. + eventually(timeout(10.seconds), interval(10.milliseconds)) { + assert(statusStore.listener.get.noLiveData()) + } } } } From cc0070c24b981e092be0126dea69830cc96ec75b Mon Sep 17 00:00:00 2001 From: Hyukjin Kwon Date: Fri, 26 Jun 2026 19:56:55 +0900 Subject: [PATCH 2/2] [SQL][TEST] Correct flake root-cause comment and reduce eventually timeout Address review feedback: the doAsync metrics-aggregation explanation was wrong under ASYNC_TRACKING_ENABLED=false (it runs inline). The real race is a trailing SparkListenerJobEnd for a failed job posted after the job waiter is notified, so it can arrive after waitUntilEmpty() drains the bus. Reduce the eventually timeout from 10s to 5s to avoid doubling the worst-case hang. Co-authored-by: Isaac --- .../execution/ui/SQLAppStatusListenerSuite.scala | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 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 179ba251754e6..6af000519aadf 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 @@ -1120,11 +1120,17 @@ 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. The cleanup of live - // executions/stage metrics is finalized when the metrics aggregation triggered by the - // SQLExecutionEnd event completes, so wait for the listener to drain rather than asserting - // immediately to avoid a timing race. - eventually(timeout(10.seconds), interval(10.milliseconds)) { + // 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()) } }