[SPARK-56565][CONNECT][TESTS] Retry flaky python foreachBatch termination test

### What changes were proposed in this pull request?
This PR un-skips the Connect parity test `test_streaming_foreach_batch_graceful_stop` and applies the retry-timeout infrastructure introduced in SPARK-52843 (`eventually` and `timeout`). Additionally, it replaces the py4j-specific `_jvm.java.lang.Thread.sleep()` call with standard Python `time.sleep()`.

### Why are the changes needed?
This resolves [SPARK-56565](https://issues.apache.org/jira/browse/SPARK-56565). The test for `foreachBatch` graceful stop was previously skipped for Connect because it inherently relied on Py4J's pinned thread execution model and caused flaky timeout behaviors. With the new retry infrastructure, we can safely re-enable it for the Connect module.

### Does this PR introduce _any_ user-facing change?
No.

### How was this patch tested?
Verified by standard GitHub Actions on personal fork.

### Was this patch authored or co-authored using generative AI tooling?
No.

Closes #55481 from Feelitx/master.

Lead-authored-by: Felix <160706963+Feelitx@users.noreply.github.com>
Co-authored-by: Luqina <cuong.m9325@gmail.com>
Signed-off-by: Hyukjin Kwon <gurwls223@apache.org>

Author: 2 people

python/pyspark/sql/tests/connect/streaming/test_parity_foreach_batch.py

@@ -15,10 +15,10 @@
 # limitations under the License.
 #
 
-import unittest
-
+import time
 from pyspark.sql.tests.streaming.test_streaming_foreach_batch import StreamingTestsForeachBatchMixin
 from pyspark.testing.connectutils import ReusedConnectTestCase, should_test_connect
+from pyspark.testing.utils import eventually, timeout
 from pyspark.errors import PySparkPicklingError
 
 if should_test_connect:
@@ -29,9 +29,17 @@ class StreamingForeachBatchParityTests(StreamingTestsForeachBatchMixin, ReusedCo
     def test_streaming_foreach_batch_propagates_python_errors(self):
         super().test_streaming_foreach_batch_propagates_python_errors()
 
-    @unittest.skip("This seems specific to py4j and pinned threads. The intention is unclear")
+    @eventually(timeout=180, catch_timeout=True)
+    @timeout(timeout=60)
     def test_streaming_foreach_batch_graceful_stop(self):
-        super().test_streaming_foreach_batch_graceful_stop()
+        # SPARK-39218: Make foreachBatch streaming query stop gracefully
+        def func(batch_df, _):
+            time.sleep(10)
+
+        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.")
 
     def test_nested_dataframes(self):
         def curried_function(df):