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 <[email protected]>
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 <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
---
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: [email protected]
For additional commands, e-mail: [email protected]
