[incubator-celeborn] 20/42: [CELEBORN-381][FLINK] notify the task with the error message when channel in active. (#1341)

[re...@apache.org]

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

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

commit faaf65f43785cea8b86aa29e5a4524e49efe7615
Author: Shuang <lv...@gmail.com>
AuthorDate: Tue Mar 14 11:28:03 2023 +0800

    [CELEBORN-381][FLINK] notify the task with the error message when channel in active. (#1341)
---
 .../plugin/flink/network/ReadClientHandler.java        | 11 ++++++++++-
 .../plugin/flink/readclient/RssBufferStream.java       |  2 +-
 .../common/network/protocol/TransportableError.java    |  6 ++++--
 .../apache/celeborn/common/util/ExceptionUtils.java    | 18 ++++++++++++++++++
 4 files changed, 33 insertions(+), 4 deletions(-)

diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
index ed6a87015..cdcccceff 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/network/ReadClientHandler.java
@@ -17,6 +17,7 @@
 
 package org.apache.celeborn.plugin.flink.network;
 
+import java.nio.charset.StandardCharsets;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
 
@@ -103,7 +104,15 @@ public class ReadClientHandler extends BaseMessageHandler {
     streamClients.forEach(
         (streamId, savedClient) -> {
           if (savedClient == client) {
-            logger.warn("Client {} is lost, remove related stream {}", savedClient, streamId);
+            String message =
+                "Client "
+                    + client.getSocketAddress()
+                    + " is lost, notify related stream "
+                    + streamId;
+            logger.warn(message);
+            processMessageInternal(
+                streamId,
+                new TransportableError(streamId, message.getBytes(StandardCharsets.UTF_8)));
           }
         });
   }
diff --git a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
index 16cc03419..3383a6439 100644
--- a/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
+++ b/client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/readclient/RssBufferStream.java
@@ -94,7 +94,7 @@ public class RssBufferStream {
 
           @Override
           public void onFailure(Throwable e) {
-            throw new RuntimeException("OpenStream failed.", e);
+            messageConsumer.accept(new TransportableError(streamId, e));
           }
         });
   }
diff --git a/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportableError.java b/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportableError.java
index 628bf8509..8762067c1 100644
--- a/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportableError.java
+++ b/common/src/main/java/org/apache/celeborn/common/network/protocol/TransportableError.java
@@ -20,7 +20,8 @@ package org.apache.celeborn.common.network.protocol;
 import java.nio.charset.StandardCharsets;
 
 import io.netty.buffer.ByteBuf;
-import org.apache.commons.lang3.exception.ExceptionUtils;
+
+import org.apache.celeborn.common.util.ExceptionUtils;
 
 public class TransportableError extends RequestMessage {
   private long streamId;
@@ -28,7 +29,8 @@ public class TransportableError extends RequestMessage {
 
   public TransportableError(long streamId, Throwable throwable) {
     this.streamId = streamId;
-    this.errorMessage = ExceptionUtils.getStackTrace(throwable).getBytes(StandardCharsets.UTF_8);
+    this.errorMessage =
+        ExceptionUtils.stringifyException(throwable).getBytes(StandardCharsets.UTF_8);
   }
 
   public TransportableError(long streamId, byte[] errorMessage) {
diff --git a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java b/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
index 119351d2d..ea218122d 100644
--- a/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
+++ b/common/src/main/java/org/apache/celeborn/common/util/ExceptionUtils.java
@@ -18,6 +18,8 @@
 package org.apache.celeborn.common.util;
 
 import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
 
 import org.apache.celeborn.common.exception.CelebornIOException;
 
@@ -32,4 +34,20 @@ public class ExceptionUtils {
       throw new CelebornIOException(exception.getMessage(), exception);
     }
   }
+
+  public static String stringifyException(Throwable exception) {
+    if (exception == null) {
+      return "(null)";
+    }
+
+    try {
+      StringWriter stm = new StringWriter();
+      PrintWriter wrt = new PrintWriter(stm);
+      exception.printStackTrace(wrt);
+      wrt.close();
+      return stm.toString();
+    } catch (Throwable throwable) {
+      return exception.getClass().getName() + " (error while printing stack trace)";
+    }
+  }
 }