This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b628a945a242 [SPARK-59985][ML][PYTHON] Fix malformed warning in
TorchDistributor w…
b628a945a242 is described below
commit b628a945a2423c2ce8ec23f3a19fd9e287ec5a14
Author: Firas Bouzazi <[email protected]>
AuthorDate: Sat Oct 10 17:54:15 2026 -0700
[SPARK-59985][ML][PYTHON] Fix malformed warning in TorchDistributor w…
### What changes were proposed in this pull request?
Fix the warning that `TorchDistributor._run_distributed_training` emits
when the log streaming server fails to start. The message had a stray comma, so
the error text was passed to `logger.warning` as an extra positional argument.
Formatting the record then raised `TypeError: not all arguments converted
during string formatting`. The warning now uses a single format string with the
exception as its argument (`"... error: %r.", e`).
### Why are the changes needed?
The fallback added in SPARK-44909 still works (training continues with the
port set to `-1`), but the warning that tells users why worker logs are missing
is broken:
- With default logging settings, users see a `--- Logging error ---`
traceback that looks like a crash instead of a one-line warning.
- With `logging.raiseExceptions = False`, the warning is silently dropped.
- `record.getMessage()` raises, so the warning cannot be asserted on.
### Does this PR introduce _any_ user-facing change?
Yes. When the log streaming server fails to start, users now get the
intended warning instead of a logging error traceback.
Before:
```
--- Logging error ---
Traceback (most recent call last):
...
TypeError: not all arguments converted during string formatting
Call stack:
File ".../pyspark/ml/torch/distributor.py", line 815, in
_run_distributed_training
self.logger.warning(
Message: 'Start torch distributor log streaming server failed, You cannot
receive logs sent from distributor workers, '
Arguments: ("error: OSError('Address already in use').",)
```
After:
```
Start torch distributor log streaming server failed, You cannot receive
logs sent from distributor workers, error: OSError('Address already in use').
```
### How was this patch tested?
Added `test_log_streaming_server_start_failure_warns` to
`TorchDistributorBaselineUnitTestsMixin`. It makes `LogStreamingServer.start`
raise and checks that execution reaches task-function creation with log
streaming disabled and that exactly one warning is logged containing the error.
The test fails with the `TypeError` above before the fix and passes after it.
The rest of `TorchDistributorBaselineUnitTests` still passes.
```
python/run-tests --testnames 'pyspark.ml.torch.tests.test_distributor
TorchDistributorBaselineUnitTests'
```
### Was this patch authored or co-authored using generative AI tooling?
Co Authered with claude.
Closes #59242 from firasbouzazi/SPARK-59985-fix-torch-distributor-warning.
Authored-by: Firas Bouzazi <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit d7635d099d6d2c71b242d5549ae64846076287d5)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
python/pyspark/ml/torch/distributor.py | 5 +++--
python/pyspark/ml/torch/tests/test_distributor.py | 25 +++++++++++++++++++++++
2 files changed, 28 insertions(+), 2 deletions(-)
diff --git a/python/pyspark/ml/torch/distributor.py
b/python/pyspark/ml/torch/distributor.py
index 49fd800db99e..3c2d99363921 100644
--- a/python/pyspark/ml/torch/distributor.py
+++ b/python/pyspark/ml/torch/distributor.py
@@ -809,8 +809,9 @@ class TorchDistributor(Distributor):
self.log_streaming_server_port = -1
self.logger.warning(
"Start torch distributor log streaming server failed, "
- "You cannot receive logs sent from distributor workers, ",
- f"error: {repr(e)}.",
+ "You cannot receive logs sent from distributor workers, "
+ "error: %r.",
+ e,
)
try:
diff --git a/python/pyspark/ml/torch/tests/test_distributor.py
b/python/pyspark/ml/torch/tests/test_distributor.py
index 5d5701f0d04a..af994c81af40 100644
--- a/python/pyspark/ml/torch/tests/test_distributor.py
+++ b/python/pyspark/ml/torch/tests/test_distributor.py
@@ -275,6 +275,31 @@ class TorchDistributorBaselineUnitTestsMixin:
)
self.assertIn("hello_after_socket_drop", output.getvalue().strip())
+ def test_log_streaming_server_start_failure_warns(self) -> None:
+ """A failure to start the log streaming server must only emit a
warning."""
+
+ class StopTraining(Exception):
+ pass
+
+ dist = TorchDistributor(num_processes=2, local_mode=False,
use_gpu=False)
+ with (
+ patch(
+ "pyspark.ml.torch.distributor.LogStreamingServer.start",
+ side_effect=OSError("Address already in use"),
+ ),
+ patch.object(dist, "_get_spark_task_function",
side_effect=StopTraining),
+ self.assertLogs(dist.logger, level="WARNING") as logs,
+ self.assertRaises(StopTraining),
+ ):
+ dist._run_distributed_training(MagicMock(), MagicMock(), None,
None)
+
+ self.assertEqual(dist.log_streaming_server_port, -1)
+ self.assertIsNone(dist.log_streaming_auth_secret)
+ self.assertEqual(len(logs.records), 1)
+ message = logs.records[0].getMessage()
+ self.assertIn("Start torch distributor log streaming server failed",
message)
+ self.assertIn("Address already in use", message)
+
def test_create_torchrun_command(self) -> None:
train_path = "train.py"
args_string = ["1", "3"]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
