This is an automated email from the ASF dual-hosted git repository.
turboFei pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new efa3c4c3e7 [CELEBORN-2402] Log the hostname for celeborn shuffle fetch
failures
efa3c4c3e7 is described below
commit efa3c4c3e77330b2637371b1dbab5f0df5c77502
Author: r7raul1984 <[email protected]>
AuthorDate: Thu Aug 6 13:32:08 2026 +0800
[CELEBORN-2402] Log the hostname for celeborn shuffle fetch failures
### What changes were proposed in this pull request?
Added `workerHost` parameter to the
`ExceptionMaker.makeFetchFailureException()` interface and all its
implementations (spark-3). `CelebornInputStream` now resolves the hostname from
`currentReader.getLocation()` (falling back to `"unknown"` if unavailable) and
embeds it in the `FetchFailedException` message.
New exception message format:
```
Celeborn FetchFailure with appShuffleId/shuffleId: <id>/<id> from worker
<host>
```
### Why are the changes needed?
Previously, the `FetchFailedException` message contained no information
about which worker was involved in the failure. This made it difficult to
identify the problematic worker during incident investigation. Embedding the
hostname improves observability at no cost.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
Added unit test in `CelebornShuffleReaderSuite` (spark-3) to verify that
the worker hostname appears in the `FetchFailedException` message.
Closes #3780 from r7raul1984/CELEBORN-2402.
Authored-by: r7raul1984 <[email protected]>
Signed-off-by: Nicholas Jiang <[email protected]>
(cherry picked from commit e8464d6461a4eb37f17790c39f2fdbf4132ad941)
Signed-off-by: Fei Wang <[email protected]>
---
.../org/apache/spark/celeborn/ExceptionMakerHelper.scala | 3 ++-
.../spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala | 9 +++++++++
.../org/apache/celeborn/client/read/CelebornInputStream.java | 12 ++++++++++--
.../java/org/apache/celeborn/common/util/ExceptionMaker.java | 2 +-
.../celeborn/tests/spark/CelebornFetchFailureSuite.scala | 1 +
5 files changed, 23 insertions(+), 4 deletions(-)
diff --git
a/client-spark/spark-3/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala
b/client-spark/spark-3/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala
index 72d9019f8e..9f9dff6578 100644
---
a/client-spark/spark-3/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala
+++
b/client-spark/spark-3/src/main/scala/org/apache/spark/celeborn/ExceptionMakerHelper.scala
@@ -30,6 +30,7 @@ object ExceptionMakerHelper {
appShuffleId: Int,
shuffleId: Int,
partitionId: Int,
+ workerHost: String,
e: Exception): Exception = {
new FetchFailedException(
null,
@@ -37,7 +38,7 @@ object ExceptionMakerHelper {
-1,
-1,
partitionId,
- FETCH_FAILURE_ERROR_MSG + appShuffleId + "/" + shuffleId,
+ FETCH_FAILURE_ERROR_MSG + appShuffleId + "/" + shuffleId + " from
worker " + workerHost,
e)
}
}
diff --git
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
index acd68bfff7..c8ac70c59b 100644
---
a/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
+++
b/client-spark/spark-3/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleReaderSuite.scala
@@ -24,6 +24,7 @@ import java.util.concurrent.{CountDownLatch,
TimeoutException, TimeUnit}
import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger,
AtomicReference}
import org.apache.spark.{Dependency, ShuffleDependency, TaskContext}
+import org.apache.spark.celeborn.ExceptionMakerHelper
import org.apache.spark.shuffle.ShuffleReadMetricsReporter
import org.mockito.ArgumentMatchers.any
import org.mockito.Mockito
@@ -342,6 +343,14 @@ class CelebornShuffleReaderSuite extends AnyFunSuite {
}
}
+ test("FetchFailureException message includes workerHost") {
+ val workerHost = "celeborn-worker-1"
+ val e = ExceptionMakerHelper.SHUFFLE_FETCH_FAILURE_EXCEPTION_MAKER
+ .makeFetchFailureException(1, 2, 3, workerHost, new IOException("test"))
+ assert(e.getMessage.contains(workerHost))
+ assert(e.getMessage.contains(ExceptionMakerHelper.FETCH_FAILURE_ERROR_MSG))
+ }
+
private def newLocation(id: Int, host: String, fetchPort: Int):
PartitionLocation =
new PartitionLocation(id, 0, host, 0, 0, fetchPort, 0,
PartitionLocation.Mode.PRIMARY)
}
diff --git
a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
index 614ce5926f..e9dc4b9625 100644
---
a/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
+++
b/client/src/main/java/org/apache/celeborn/client/read/CelebornInputStream.java
@@ -991,14 +991,22 @@ public abstract class CelebornInputStream extends
InputStream {
if (exceptionMaker != null) {
if (shuffleClient.reportShuffleFetchFailure(appShuffleId, shuffleId,
taskId)) {
/*
- * [[ExceptionMaker.makeException]], for spark applications with
celeborn.client.spark.stageRerun.enabled enabled will result in creating
+ * [[ExceptionMaker.makeFetchFailureException]], for spark
applications with celeborn.client.spark.stageRerun.enabled enabled will result
in creating
* a FetchFailedException; and that will make the TaskContext as
failed with shuffle fetch issues - see SPARK-19276 for more.
* Given this, Celeborn can wrap the FetchFailedException with our
CelebornIOException
*/
+ String workerHost =
+ Optional.ofNullable(currentReader)
+ .map(
+ r ->
+ Optional.ofNullable(r.getLocation())
+ .map(l -> l.getHost())
+ .orElse("unknown"))
+ .orElse("unknown");
ioe =
new CelebornIOException(
exceptionMaker.makeFetchFailureException(
- appShuffleId, shuffleId, partitionId, e));
+ appShuffleId, shuffleId, partitionId, workerHost, e));
}
}
throw ioe;
diff --git
a/common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java
b/common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java
index 4f08ff3f50..6dde6ae2f3 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java
+++ b/common/src/main/java/org/apache/celeborn/common/util/ExceptionMaker.java
@@ -19,5 +19,5 @@ package org.apache.celeborn.common.util;
public interface ExceptionMaker {
Exception makeFetchFailureException(
- int appShuffleId, int shuffleId, int partitionId, Exception e);
+ int appShuffleId, int shuffleId, int partitionId, String workerHost,
Exception e);
}
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
index 58cfe27594..49558dfe6f 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/CelebornFetchFailureSuite.scala
@@ -319,6 +319,7 @@ class CelebornFetchFailureSuite extends AnyFunSuite
appShuffleId,
-1,
context.partitionId(),
+ "unknown",
new IOException("forced"))
}
iter
