code: def test_listener_events(self):
      start_event = None
      progress_event = None
      terminated_event = None
  
      # V1: Initial interface of StreamingQueryListener containing methods `onQueryStarted`,
      # `onQueryProgress`, `onQueryTerminated`. It is prior to Spark 3.5.
      class TestListenerV1(StreamingQueryListener):
          def onQueryStarted(self, event):
              nonlocal start_event
              start_event = event
  
          def onQueryProgress(self, event):
              nonlocal progress_event
              progress_event = event
  
          def onQueryTerminated(self, event):
              nonlocal terminated_event
              terminated_event = event
  
      # V2: The interface after the method `onQueryIdle` is added. It is Spark 3.5+.
      class TestListenerV2(StreamingQueryListener):
          def onQueryStarted(self, event):
              nonlocal start_event
              start_event = event
  
          def onQueryProgress(self, event):
              nonlocal progress_event
              progress_event = event
  
          def onQueryIdle(self, event):
              pass
  
          def onQueryTerminated(self, event):
              nonlocal terminated_event
              terminated_event = event
  
      def verify(test_listener):
          nonlocal start_event
          nonlocal progress_event
          nonlocal terminated_event
  
          start_event = None
          progress_event = None
          terminated_event = None
  
          try:
              self.spark.streams.addListener(test_listener)
  
              df = (
                  self.spark.readStream.format("rate")
                  .option("rowsPerSecond", 10)
                  .load()
              )
  
              # check successful stateful query
              df_stateful = df.groupBy().count()  # make query stateful
              q = (
                  df_stateful.writeStream.format("noop")
                  .queryName("test")
                  .outputMode("complete")
                  .start()
              )
              self.assertTrue(q.isActive)
              time.sleep(10)
              q.stop()
  
              # Make sure all events are empty
              self.spark.sparkContext._jsc.sc().listenerBus().waitUntilEmpty()
  
              self.check_start_event(start_event)
              self.check_progress_event(progress_event)
              self.check_terminated_event(terminated_event)
  
              # Check query terminated with exception
              from pyspark.sql.functions import col, udf
  
              bad_udf = udf(lambda x: 1 / 0)
              q = (
                  df.select(bad_udf(col("value")))
                  .writeStream.format("noop")
                  .start()
              )
              time.sleep(5)
              q.stop()
              self.spark.sparkContext._jsc.sc().listenerBus().waitUntilEmpty()
              self.check_terminated_event(
                  terminated_event, "ZeroDivisionError"
              )
  
          finally:
              self.spark.streams.removeListener(test_listener)
  
      verify(TestListenerV1())
      verify(TestListenerV2())
  
imports: import json
  import time
  import uuid
  from datetime import datetime
  from pyspark import Row
  from pyspark.sql.streaming import StreamingQueryListener
  from pyspark.sql.streaming.listener import QueryStartedEvent, QueryProgressEvent, QueryTerminatedEvent, SinkProgress, SourceProgress, StateOperatorProgress, StreamingQueryProgress
  from pyspark.testing.sqlutils import ReusedSQLTestCase
label_desc: Use image node_id 2 for calling the StreamingListenerTests obj's underlying member method code with example usage: obj.test_listener_events() without return types