You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ga...@apache.org on 2026/07/13 18:43:21 UTC
(spark) branch branch-4.1 updated: [SPARK-58020][4.1][PYTHON][TEST] Fix flaky test_data_source_writer_with_logging
This is an automated email from the ASF dual-hosted git repository.
gaogaotiantian pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new be18fadfba0d [SPARK-58020][4.1][PYTHON][TEST] Fix flaky test_data_source_writer_with_logging
be18fadfba0d is described below
commit be18fadfba0d60087b03f51804d736ae07d69ee3
Author: Tian Gao <ga...@hotmail.com>
AuthorDate: Mon Jul 13 11:43:08 2026 -0700
[SPARK-58020][4.1][PYTHON][TEST] Fix flaky test_data_source_writer_with_logging
### What changes were proposed in this pull request?
This is a test-only fix that isolates the flake fix from SPARK-55799 for `branch-4.1`. `test_data_source_writer_with_logging` asserted exactly two `"TestJsonWriter.write: abort test"` log rows. That count is not guaranteed: when the first partition is aborted, the executor cancels the remaining tasks, so whether the second partition logs before cancellation is time sensitive. The assertion is split so abort logs are matched with `dropDuplicates(["msg"])`, tolerating either 1 or 2 abort logs.
Note: SPARK-55799 on master also changed `python/pyspark/logger/worker_io.py` and added a `PythonDataSourceTestsWithSimpleWorker` suite. Those are intentionally **not** included here -- the simple-worker suite depends on the `worker_io.py` change, and this PR is scoped to just de-flaking the existing test on `branch-4.1`.
### Why are the changes needed?
The test is flaky on `branch-4.1` (e.g. [this run](https://github.com/apache/spark/actions/runs/28858827107/job/85596316246) failed with `[DIFFERENT_ROWS]`, missing the second abort log). master and branch-4.2 already carry the fix via SPARK-55799; branch-4.1 was never backported.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing `test_data_source_writer_with_logging` in `pyspark.sql.tests.test_python_datasource`. The revised assertion passes whether the race produces 1 or 2 abort logs.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
Closes #57100 from gaogaotiantian/SPARK-58020-flaky-datasource-writer-logging.
Authored-by: Tian Gao <ga...@hotmail.com>
Signed-off-by: Tian Gao <ga...@hotmail.com>
---
python/pyspark/sql/tests/test_python_datasource.py | 33 ++++++++++++++++------
1 file changed, 24 insertions(+), 9 deletions(-)
diff --git a/python/pyspark/sql/tests/test_python_datasource.py b/python/pyspark/sql/tests/test_python_datasource.py
index e5171876656c..0f5768dbd914 100644
--- a/python/pyspark/sql/tests/test_python_datasource.py
+++ b/python/pyspark/sql/tests/test_python_datasource.py
@@ -1199,8 +1199,20 @@ class BasePythonDataSourceTestsMixin:
logs = self.spark.tvf.python_worker_logs()
+ # We could get either 1 or 2 "TestJsonWriter.write: abort test" logs because
+ # the operation is time sensitive. When the first partition gets aborted,
+ # the executor will cancel the rest of the tasks. Whether we are able to get
+ # the second log depends on whether the second partition starts before the
+ # cancellation. When we use simple worker, the second log is often missing
+ # because the spawn overhead is large.
+ non_abort_logs = logs.select("level", "msg", "context", "logger").filter(
+ "msg != 'TestJsonWriter.write: abort test'"
+ )
+ abort_logs = logs.select("level", "msg", "context", "logger").filter(
+ "msg == 'TestJsonWriter.write: abort test'"
+ )
assertDataFrameEqual(
- logs.select("level", "msg", "context", "logger"),
+ non_abort_logs,
[
Row(
level="WARNING",
@@ -1245,14 +1257,6 @@ class BasePythonDataSourceTestsMixin:
"TestJsonWriter.__init__: ['abort', 'path']",
{"class_name": "TestJsonDataSource", "func_name": "writer"},
),
- (
- "TestJsonWriter.write: abort test",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
- (
- "TestJsonWriter.write: abort test",
- {"class_name": "TestJsonWriter", "func_name": "write"},
- ),
(
"TestJsonWriter.abort",
{"class_name": "TestJsonWriter", "func_name": "abort"},
@@ -1260,6 +1264,17 @@ class BasePythonDataSourceTestsMixin:
]
],
)
+ assertDataFrameEqual(
+ abort_logs.dropDuplicates(["msg"]),
+ [
+ Row(
+ level="WARNING",
+ msg="TestJsonWriter.write: abort test",
+ context={"class_name": "TestJsonWriter", "func_name": "write"},
+ logger="test_datasource_writer",
+ )
+ ],
+ )
class PythonDataSourceTests(BasePythonDataSourceTestsMixin, ReusedSQLTestCase):
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org