PySpark job doesnt stop on stopping query

[ronban]

Bug Report:

Scenario:

...
ehConf = {}
ehConf['eventhubs.useExclusiveReceiver'] = False
ehConf['eventhubs.connectionString'] = spark.sparkContext._jvm.org.apache.spark.eventhubs.EventHubsUtils.encrypt(
        connectionString)
startingEventPosition = {
        "offset": "-1",
        "seqNo": -1,  # not in use
        "enqueuedTime": None,  # not in use
        "isInclusive": True
}
endingEventPosition = {
        "offset": "@latest",
        "seqNo": -1,
        "enqueuedTime": None,
        "isInclusive": True
}
ehConf["eventhubs.startingPosition"] = dumps(startingEventPosition)
ehConf["eventhubs.endingPosition"] = dumps(endingEventPosition)


qs=spark.readStream.format("eventhubs")\
            .options(**ehConf)\
            .load()\
            .writeStream\
            .foreachBatch(lambda bdf,bid: my_func(bdf,bid,schema,storage_format,coalesced_output))\
            .option("checkpointLocation", checkpoint)\
            .start()

    monitor_thread = threading.Thread(target=monitor,args=(qs,))
    monitor_thread.start()
    monitor_thread.join()

    qs.awaitTermination()

def monitor(qs:StreamingQuery):
    time.sleep(10)
    while True:
        if not qs.status.get('isDataAvailable'):
            print("Stopping Query Stream")
            qs.stop()
            break
        else:
            time.sleep(1)


...
sparkSession.stop()

• Actual behavior: For other types of streams for e.g. kafka, this stops, for eventhubs, this doesnt
• Expected behavior: The job should end successfully
• Spark version: Dataproc batches v 1.1.2
• spark-eventhubs artifactId and version : com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.22