[SPARK-57711][SQL][TEST] Deflake SQLAppStatusListenerMemoryLeakSuite "no memory leak"

[HyukjinKwon]

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:

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.

[uros-b]

The added comment / PR body pin the flake on the kvstore.doAsync cleanup completing after the bus drains, but the test sets ASYNC_TRACKING_ENABLED=false, which makes doAsync run synchronously on the listener thread (sameThreadExecutor); so waitUntilEmpty() should already cover that path. The stated root cause likely doesn't hold under this config; the real residual (e.g. stageMetrics cleanup) isn't identified. The fix is still safe, but please clarify the actual race (or confirm the CI env doesn't apply the config) so this isn't masking an undiagnosed condition. Tighten the inline comment accordingly.

[HyukjinKwon]

Good catch, you're right. With ASYNC_TRACKING_ENABLED=false the kvstore.doAsync block in onExecutionEnd runs inline (ElementTrackingStore uses sameThreadExecutorService), so it's already covered by waitUntilEmpty() and isn't the cause.

The actual race is in end-event delivery. The test runs failing jobs (df.foreach { throw ... }), and for a failed job DAGScheduler.failJobAndIndependentStages notifies the job waiter -- unblocking the failing action on the test thread -- before it posts SparkListenerJobEnd:

cleanupStateForJobAndIndependentStages(job)
job.listener.jobFailed(error)                                          // unblocks the action
listenerBus.post(SparkListenerJobEnd(job.jobId, ..., JobFailed(error))) // posted afterwards

So the test thread can race ahead, exit the loop, and call waitUntilEmpty() before the trailing JobEnd for the last failed execution is enqueued. If it lands just after the bus drains, that execution never reaches the jobs.size + 1 cleanup threshold and lingers in liveExecutions, so noLiveData() is intermittently false. I've rewritten the inline comment and PR description accordingly.

[uros-b]

Note: stacking a 10s eventually on top of the existing 10s waitUntilEmpty() doubles worst-case hang to ~20s on a genuinely broken system.

[HyukjinKwon]

Reduced the eventually timeout to 5.seconds. The trailing end event lands within milliseconds in practice, so the worst-case additive hang is now ~15s rather than ~20s.