HyukjinKwon commented on code in PR #56790:
URL: https://github.com/apache/spark/pull/56790#discussion_r3480926398
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala:
##########
@@ -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
Review Comment:
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`:
```scala
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.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala:
##########
@@ -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)) {
Review Comment:
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.
--
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]
