@@ -373,6 +373,7 @@ class CometIcebergRewriteActionSuite extends CometTestBase with CometIcebergTest
         case _ =>
       }
     }
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.sparkContext.addSparkListener(listener)
     try {
       body
 
@@ -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 =
 
@@ -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
 
@@ -529,6 +529,7 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {
           qe: QueryExecution,
           exception: Exception): Unit = {}
     }
+    CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
     spark.listenerManager.register(listener)
     try {
       action
Original file line number	Diff line number	Diff line change
`@@ -373,6 +373,7 @@ class CometIcebergRewriteActionSuite extends CometTestBase with CometIcebergTest`
`373`	`373`	`case _ =>`
`374`	`374`	`}`
`375`	`375`	`}`
	`376`	`+ CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)`
`376`	`377`	`spark.sparkContext.addSparkListener(listener)`
`377`	`378`	`try {`
`378`	`379`	`body`
Original file line number	Diff line number	Diff line change
`@@ -529,6 +529,7 @@ object CometIcebergWriteBenchmark extends CometBenchmarkBase {`
`529`	`529`	`qe: QueryExecution,`
`530`	`530`	`exception: Exception): Unit = {}`
`531`	`531`	`}`
	`532`	`+ CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)`
`532`	`533`	`spark.listenerManager.register(listener)`
`533`	`534`	`try {`
`534`	`535`	`action`