(incubator-celeborn) branch branch-0.4 updated: [CELEBORN-1253] Improve exception message of fetching chunk failure for WorkerPartitionReader

[nicholasjiang]

This is an automated email from the ASF dual-hosted git repository.

nicholasjiang pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.4 by this push:
     new 1f387b966 [CELEBORN-1253] Improve exception message of fetching chunk 
failure for WorkerPartitionReader
1f387b966 is described below

commit 1f387b96688ae169f4a3d80211bf7534a1a78c7e
Author: SteNicholas <programg...@163.com>
AuthorDate: Thu Jan 25 11:48:29 2024 +0800

    [CELEBORN-1253] Improve exception message of fetching chunk failure for 
WorkerPartitionReader
    
    ### What changes were proposed in this pull request?
    
    Improve exception message of fetching chunk failure for 
`WorkerPartitionReader` including shuffle key.
    
    ### Why are the changes needed?
    
    The exception message of fetching chunk failure for `WorkerPartitionReader` 
does not contain shuffle key of the chunk, which could not match the log of 
Worker to troubleshooting. It's recommend to add shuffle key in exception 
message of fetching chunk failure.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    No.
    
    Closes #2261 from SteNicholas/CELEBORN-1253.
    
    Authored-by: SteNicholas <programg...@163.com>
    Signed-off-by: SteNicholas <programg...@163.com>
    (cherry picked from commit 8c0b434ffa0523e043d371158b98c51f398f526a)
    Signed-off-by: SteNicholas <programg...@163.com>
---
 .../org/apache/celeborn/client/read/WorkerPartitionReader.java    | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git 
a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
 
b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
index a00f02476..0d474ee72 100644
--- 
a/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
+++ 
b/client/src/main/java/org/apache/celeborn/client/read/WorkerPartitionReader.java
@@ -60,6 +60,7 @@ public class WorkerPartitionReader implements PartitionReader 
{
   private final ChunkReceivedCallback callback;
 
   private final AtomicReference<IOException> exception = new 
AtomicReference<>();
+  private final String shuffleKey;
   private final int fetchMaxReqsInFlight;
   private final long fetchTimeoutMs;
   private boolean closed = false;
@@ -80,6 +81,7 @@ public class WorkerPartitionReader implements PartitionReader 
{
       int fetchChunkMaxRetry,
       MetricsCallback metricsCallback)
       throws IOException, InterruptedException {
+    this.shuffleKey = shuffleKey;
     fetchMaxReqsInFlight = conf.clientFetchMaxReqsInFlight();
     results = new LinkedBlockingQueue<>();
     fetchTimeoutMs = conf.clientFetchTimeoutMs();
@@ -101,7 +103,8 @@ public class WorkerPartitionReader implements 
PartitionReader {
 
           @Override
           public void onFailure(int chunkIndex, Throwable e) {
-            String errorMsg = "Fetch chunk " + chunkIndex + " failed.";
+            String errorMsg =
+                String.format("Fetch chunk %d of shuffle key %s failed.", 
chunkIndex, shuffleKey);
             logger.error(errorMsg, e);
             exception.set(new CelebornIOException(errorMsg, e));
           }
@@ -207,7 +210,8 @@ public class WorkerPartitionReader implements 
PartitionReader {
               client = clientFactory.createClient(location.getHost(), 
location.getFetchPort());
             } catch (IOException e) {
               logger.error(
-                  "fetchChunk for streamId: {}, chunkIndex: {} failed.",
+                  "FetchChunk for shuffleKey: {}, streamId: {}, chunkIndex: {} 
failed.",
+                  shuffleKey,
                   streamHandler.getStreamId(),
                   chunkIndex,
                   e);