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 814fee1343d0 [SPARK-59985][4.1][ML][PYTHON] Fix malformed warning in
TorchDistributor when the log streaming server fails to start
814fee1343d0 is described below
commit 814fee1343d058b154c2a3762746e7a45d5827a9
Author: Firas Bouzazi <[email protected]>
AuthorDate: Wed Oct 28 16:31:52 2026 -0700
[SPARK-59985][4.1][ML][PYTHON] Fix malformed warning in TorchDistributor
when the log streaming server fails to start
### What changes were proposed in this pull request?
Backport of #59242 to branch-4.1.
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,
and formatting the record raised `TypeError: not all arguments converted during
string formatting`. The warning now uses a single format string with the
exception as its argument.
The cherry-pick conflicted only in the test file, where master has a
neighbouring test that is not on this branch. Adaptations to the test:
- The test does not assert `log_streaming_auth_secret`, which does not
exist on this branch.
- The test file now imports `MagicMock`, which it did not import on this
branch.
### Why are the changes needed?
Users see a `--- Logging error ---` traceback instead of the one-line
warning explaining why worker logs are missing (or nothing, with
`logging.raiseExceptions = False`).
### Does this PR introduce _any_ user-facing change?
Yes. When the log streaming server fails to start, users get the intended
warning instead of a logging error traceback.
### How was this patch tested?
`test_log_streaming_server_start_failure_warns` (added in #59242) fails
with the `TypeError` above without the fix and passes with it on this branch.
`TorchDistributorBaselineUnitTests` passes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code
Closes #59383 from firasbouzazi/SPARK-59985-4.1.
Authored-by: Firas Bouzazi <[email protected]>
Signed-off-by: Tian Gao <[email protected]>
---
python/pyspark/ml/torch/distributor.py | 5 +++--
python/pyspark/ml/torch/tests/test_distributor.py | 26 ++++++++++++++++++++++-
2 files changed, 28 insertions(+), 3 deletions(-)
diff --git a/python/pyspark/ml/torch/distributor.py
b/python/pyspark/ml/torch/distributor.py
index ef86f38b716b..8b2fde12335a 100644
--- a/python/pyspark/ml/torch/distributor.py
+++ b/python/pyspark/ml/torch/distributor.py
@@ -773,8 +773,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 e9bf1d784000..184fb82c7fe1 100644
--- a/python/pyspark/ml/torch/tests/test_distributor.py
+++ b/python/pyspark/ml/torch/tests/test_distributor.py
@@ -27,7 +27,7 @@ import tempfile
import threading
from typing import Callable, Dict, Any
import unittest
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
from pyspark import SparkConf, SparkContext
from pyspark.ml.torch.distributor import TorchDistributor, _get_gpus_owned
@@ -257,6 +257,30 @@ class TorchDistributorBaselineUnitTestsMixin:
error_command = ["bash", "-c", "'abc''def'"]
TorchDistributor._execute_command(error_command)
+ 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.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]
