code: def test_streaming_foreach_batch_graceful_stop(self):
      # SPARK-39218: Make foreachBatch streaming query stop gracefully
      def func(batch_df, _):
          batch_df.sparkSession._jvm.java.lang.Thread.sleep(10000)
  
      q = (
          self.spark.readStream.format("rate")
          .load()
          .writeStream.foreachBatch(func)
          .start()
      )
      time.sleep(
          3
      )  # 'rowsPerSecond' defaults to 1. Waits 3 secs out for the input.
      q.stop()
      self.assertIsNone(
          q.exception(), "No exception has to be propagated."
      )
  
file_path: spark/python/pyspark/sql/tests/streaming/test_streaming_foreach_batch.py
imports: import time
  from pyspark.sql.dataframe import DataFrame
  from pyspark.testing.sqlutils import ReusedSQLTestCase
label_desc: Use image node_id 4 for calling the StreamingTestsForeachBatchMixin obj's underlying member method code with example usage: obj.test_streaming_foreach_batch_graceful_stop() without return types