adressing onQueryIdle on StreamingQueryListener in Spark 3.5

lawofcycles · lawofcycles · commit 76e71839d9d8 · 2026-03-01T20:29:04.000+09:00
Signed-off-by: Sotaro Hikita &lt;bering1814@gmail.com&gt;
diff --git a/spark/sql-35/src/itest/scala/org/opensearch/spark/sql/streaming/StreamingQueryTestHarness.scala b/spark/sql-35/src/itest/scala/org/opensearch/spark/sql/streaming/StreamingQueryTestHarness.scala
@@ -131,6 +131,13 @@ class StreamingQueryTestHarness[S <: java.io.Serializable : Encoder](val sparkSe
       }
     }
 
+    override def onQueryIdle(event: StreamingQueryListener.QueryIdleEvent): Unit = {
+      captureQueryID(event.id)
+      if (inputsSeen >= inputsRequired) {
+        latch.countDown()
+      }
+    }
+
     override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {
       try {
         captureQueryID(event.id)

Original file line number	Diff line number	Diff line change
`@@ -131,6 +131,13 @@ class StreamingQueryTestHarness[S <: java.io.Serializable : Encoder](val sparkSe`
`131`	`131`	`}`
`132`	`132`	`}`
`133`	`133`
	`134`	`+ override def onQueryIdle(event: StreamingQueryListener.QueryIdleEvent): Unit = {`
	`135`	`+ captureQueryID(event.id)`
	`136`	`+ if (inputsSeen >= inputsRequired) {`
	`137`	`+ latch.countDown()`
	`138`	`+ }`
	`139`	`+ }`
	`140`	`+`
`134`	`141`	`override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = {`
`135`	`142`	`try {`
`136`	`143`	`captureQueryID(event.id)`