(incubator-celeborn) branch main 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 main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


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

commit 8c0b434ffa0523e043d371158b98c51f398f526a
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>
---
 .../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);