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

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())
+        }
       }
     }
   }