[SPARK-42941][SS][CONNECT][DRAFT][DO-NOT-REVIEW] Python StreamingQueryListener POC

[WweiL]

What changes were proposed in this pull request?

Add streaming query listener for python client. In this version, the code runs on the driver.

Why are the changes needed?

Continuation development of streaming connect

Does this PR introduce any user-facing change?

Yes

How was this patch tested?

For now manually:
client:

Using Python version 3.9.16 (main, Dec  7 2022 01:11:58)
Client connected to the Spark Connect server at localhost
SparkSession available as 'spark'.
>>> from pyspark.sql.connect.streaming.listener import QueryStartedEvent;from pyspark.sql.connect.streaming.listener import StreamingQueryListener;from pyspark.sql.streaming.listener import (QueryProgressEvent, QueryTerminatedEvent, QueryIdleEvent)
>>> class MyListener(StreamingQueryListener):
...     def onQueryStarted(self, event: QueryStartedEvent) -> None:
...             print("hi, event query id is: " +  event.id)
...     def onQueryProgress(self, event: QueryProgressEvent) -> None:
...             pass
...     def onQueryTerminated(self, event: QueryTerminatedEvent) -> None:
...             pass
...     def onQueryIdle(self, event: QueryIdleEvent) -> None:
...             pass
... 
>>> spark.streams.addListener(MyListener())
>>> q = spark.readStream.format("rate").load().writeStream.format("console").start()
>>> q.stop()

server(driver):

23/05/08 21:27:49 INFO MicroBatchExecution: Starting [id = 6042353a-dc77-436b-9ee5-d4d0653ec0a2, runId = 269288ae-3463-408c-bbb3-a72934a4bc1d]. Use file:/tmp/temporary-9ec9765b-b6d8-427e-bfe9-fa758b9f87b7 to store the query checkpoint.
##### Python out for query start event is: out=###### Start running onQueryStarted ######
hi, event query id is: 6042353a-dc77-436b-9ee5-d4d0653ec0a2

##### Python error for query start event is: error=
##### End processing query start event exitCode=0

from pyspark.sql.streaming.listener import StreamingQueryListener;from pyspark.sql.streaming.listener import (QueryStartedEvent, QueryProgressEvent, QueryTerminatedEvent, QueryIdleEvent)
>>> class MyListener(StreamingQueryListener):
...     def onQueryStarted(self, event: QueryStartedEvent) -> None: print("hi, event query id is: " +  str(event.id))
...     def onQueryProgress(self, event: QueryProgressEvent) -> None: pass
...     def onQueryTerminated(self, event: QueryTerminatedEvent) -> None: pass
...     def onQueryIdle(self, event: QueryIdleEvent) -> None: pass

[WweiL]

@rangadi @pengzhon-db

[WweiL]

here