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

Co-authored-by: raoraoxiong <xiongraorao@gmail.com>
Co-authored-by: Yun Tang <myasuka@live.com>

This closes #27186

Author: RocMarshal

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

@@ -68,10 +68,14 @@ private SerializedThrowable(Throwable exception, Set<Throwable> alreadySeen) {
         if (!(exception instanceof SerializedThrowable)) {
             // serialize and memoize the original message
             byte[] serialized;
-            try {
-                serialized = InstantiationUtil.serializeObject(exception);
-            } catch (Throwable t) {
-                serialized = null;
+            // introduce the synchronization here to avoid deadlock of multi thread serializing
+            // exceptions
+            synchronized (SerializedThrowable.class) {
+                try {
+                    serialized = InstantiationUtil.serializeObject(exception);
+                } catch (Throwable t) {
+                    serialized = null;
+                }
             }
             this.serializedException = serialized;
             this.cachedException = new WeakReference<>(exception);
@@ -94,7 +98,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 +108,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 +145,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,12 +19,22 @@
 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 org.junit.jupiter.api.Timeout;
+
+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 java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -179,4 +189,62 @@ void testCopySuppressed() {
                 .isInstanceOf(SerializedThrowable.class)
                 .hasMessage("java.lang.Exception: suppressed");
     }
+
+    @Test
+    void testCyclicSuppressedThrowableSerialized() {
+        SerializedThrowable serializedThrowable = new SerializedThrowable(mockThrowable());
+        assertThat(serializedThrowable).isNotNull();
+    }
+
+    @Test
+    @Timeout(value = 5, unit = TimeUnit.SECONDS)
+    void testCyclicSuppressedThrowableConcurrentSerialized() throws InterruptedException {
+        Throwable throwable = mockThrowable();
+        int threadNum = 16;
+        CountDownLatch countDownLatch = new CountDownLatch(threadNum);
+        List<Thread> threads = new ArrayList<>();
+        for (int i = 0; i < threadNum; i++) {
+            String threadName = "thread-" + i;
+            Thread t = createThread(countDownLatch, throwable, threadName);
+            t.start();
+            countDownLatch.countDown();
+            threads.add(t);
+        }
+        for (Thread thread : threads) {
+            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;
+    }
 }