test: drain the listener bus before registering Iceberg test listeners

spark/src/test/scala/org/apache/comet/CometIcebergRewriteActionSuite.scala

@@ -373,6 +373,7 @@ class CometIcebergRewriteActionSuite extends CometTestBase with CometIcebergTest
         case _ =>
       }
     }
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.sparkContext.addSparkListener(listener)
     try {
       body

spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala

@@ -237,6 +237,9 @@ trait CometIcebergTestBase { this: CometTestBase =>
       override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit =
         if (includeFailures) captured += qe.executedPlan
     }
+    // Events from earlier queries may still be queued; drain them so they do not reach the
+    // listener.
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.listenerManager.register(listener)
     try {
       action
@@ -259,6 +262,7 @@ trait CometIcebergTestBase { this: CometTestBase =>
       override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit =
         captured += qe.executedPlan
     }
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.listenerManager.register(listener)
     try {
       val error =

spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala

@@ -3411,6 +3411,8 @@ class CometIcebergWriteActionSuite
           override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit =
             if (taskEnd.reason == Success) JobAbortGate.taskFinished()
         }
+        // Drain the setup writes' task-end events first, or they can count toward the gate.
+        CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
         spark.sparkContext.addSparkListener(listener)
         try {
           val run = () =>
@@ -4309,6 +4311,8 @@ class CometIcebergWriteActionSuite
       val listener = new IcebergWriteReportListener(
         new SparkConf()
           .set(CometConf.COMET_ICEBERG_WRITE_REPORT_DIR.key, reportDir.getAbsolutePath))
+      // Drain the setup writes' events first, or they can reach the listener and be reported.
+      CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
       spark.listenerManager.register(listener)
       try {
         action

spark/src/test/scala/org/apache/spark/sql/benchmark/CometIcebergWriteBenchmark.scala

@@ -529,6 +529,7 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
           qe: QueryExecution,
           exception: Exception): Unit = {}
     }
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.listenerManager.register(listener)
     try {
       action