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