[FLINK-36746][core] Fix the deadlock bug in SerializedThrowable

Co-authored-by: raoraoxiong <[email protected]>

Author: RocMarshal

flink-core/src/main/java/org/apache/flink/util/SerializedThrowable.java

@@ -94,7 +94,7 @@ private SerializedThrowable(Throwable exception, Set<Throwable> alreadySeen) {
                 }
             }
             // mimic suppressed exceptions
-            addAllSuppressed(exception.getSuppressed());
+            this.addAllSuppressed(exception.getSuppressed(), alreadySeen);
         } else {
             // copy from that serialized throwable
             SerializedThrowable other = (SerializedThrowable) exception;
@@ -104,7 +104,7 @@ private SerializedThrowable(Throwable exception, Set<Throwable> alreadySeen) {
             this.cachedException = other.cachedException;
             this.setStackTrace(other.getStackTrace());
             this.initCause(other.getCause());
-            this.addAllSuppressed(other.getSuppressed());
+            this.addAllSuppressed(other.getSuppressed(), alreadySeen);
         }
     }
 
@@ -141,15 +141,23 @@ public String getFullStringifiedStackTrace() {
         return fullStringifiedStackTrace;
     }
 
-    private void addAllSuppressed(Throwable[] suppressed) {
+    /**
+     * Add all suppressed exceptions to this exception.
+     *
+     * @param suppressed The suppressed exceptions to add.
+     * @param alreadySeen The set of exceptions that have already been seen.
+     */
+    private void addAllSuppressed(Throwable[] suppressed, Set<Throwable> alreadySeen) {
         for (Throwable s : suppressed) {
-            SerializedThrowable serializedThrowable;
-            if (s instanceof SerializedThrowable) {
-                serializedThrowable = (SerializedThrowable) s;
-            } else {
-                serializedThrowable = new SerializedThrowable(s);
+            if (alreadySeen.add(s)) {
+                SerializedThrowable serializedThrowable;
+                if (s instanceof SerializedThrowable) {
+                    serializedThrowable = (SerializedThrowable) s;
+                } else {
+                    serializedThrowable = new SerializedThrowable(s);
+                }
+                this.addSuppressed(serializedThrowable);
             }
-            this.addSuppressed(serializedThrowable);
         }
     }
 

flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializedThrowableTest.java

@@ -19,13 +19,21 @@
 package org.apache.flink.runtime.util;
 
 import org.apache.flink.core.testutils.CommonTestUtils;
+import org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException;
 import org.apache.flink.testutils.ClassLoaderUtils;
 import org.apache.flink.util.ExceptionUtils;
 import org.apache.flink.util.InstantiationUtil;
 import org.apache.flink.util.SerializedThrowable;
 
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.fail;
@@ -179,4 +187,60 @@ void testCopySuppressed() {
                 .isInstanceOf(SerializedThrowable.class)
                 .hasMessage("java.lang.Exception: suppressed");
     }
+
+    @Test
+    void testCyclicSuppressedThrowableSerialized() {
+        SerializedThrowable serializedThrowable = new SerializedThrowable(mockThrowable());
+        assertThat(serializedThrowable).isNotNull();
+    }
+
+    @Test
+    void testCyclicSuppressedThrowableConcurrentSerialized() throws InterruptedException {
+        Throwable throwable = mockThrowable();
+        CountDownLatch countDownLatch = new CountDownLatch(2);
+        List<Thread> list = new ArrayList<>();
+        for (int i = 0; i < 2; i++) {
+            String threadName = "t" + i;
+            Thread t = createThread(countDownLatch, throwable, threadName);
+            t.start();
+            countDownLatch.countDown();
+            list.add(t);
+        }
+        for (Thread thread : list) {
+            thread.join();
+        }
+    }
+
+    private static Thread createThread(
+            CountDownLatch countDownLatch, Throwable throwable, String threadName) {
+        Thread t =
+                new Thread(
+                        () -> {
+                            try {
+                                countDownLatch.await();
+                                SerializedThrowable serializedThrowable =
+                                        new SerializedThrowable(throwable);
+                                assertThat(serializedThrowable).isNotNull();
+                            } catch (Exception e) {
+                                throw new RuntimeException(e);
+                            }
+                        });
+        t.setName(threadName);
+        return t;
+    }
+
+    private static Throwable mockThrowable() {
+        SocketAddress remoteAddr = new InetSocketAddress(80);
+        RemoteTransportException remoteTransportException =
+                new RemoteTransportException(
+                        "Connection unexpectedly closed by remote task manager '"
+                                + remoteAddr
+                                + "'. "
+                                + "This might indicate that the remote task manager was lost.",
+                        remoteAddr,
+                        new IOException("connection reset by peer."));
+        RuntimeException runtimeException = new RuntimeException(remoteTransportException);
+        remoteTransportException.addSuppressed(runtimeException);
+        return remoteTransportException;
+    }
 }