Gyulas’ comments have been addressed.

Author: rodmeneses

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtils.java

@@ -100,15 +100,15 @@ private static String getExceptionMessage(Throwable throwable, int level) {
         var msg =
                 Optional.ofNullable(throwable.getMessage())
                         .orElse(throwable.getClass().getSimpleName());
-        level++;
-        if (level == EXCEPTION_LIMIT_FOR_EVENT_MESSAGE) {
+
+        if (level >= EXCEPTION_LIMIT_FOR_EVENT_MESSAGE) {
             return msg;
         }
 
         if (throwable.getCause() == null) {
             return msg;
         } else {
-            return msg + " -> " + getExceptionMessage(throwable.getCause(), level);
+            return msg + " -> " + getExceptionMessage(throwable.getCause(), level + 1);
         }
     }
 }

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtilsTest.java

@@ -21,6 +21,8 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.io.IOException;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for {@link ExceptionUtils}. */
@@ -31,52 +33,6 @@ void testGetExceptionMessage_nullThrowable() {
         assertThat(ExceptionUtils.getExceptionMessage(null)).isNull();
     }
 
-    @Test
-    void testGetExceptionMessage_simpleException() {
-        var ex = new RuntimeException("My Exception");
-        assertThat(ExceptionUtils.getExceptionMessage(ex)).isEqualTo("My Exception");
-    }
-
-    @Test
-    void testGetExceptionMessage_simpleExceptionNullCause() {
-        var ex = new OutOfMemoryError(null);
-        assertThat(ExceptionUtils.getExceptionMessage(ex)).isEqualTo("OutOfMemoryError");
-    }
-
-    @Test
-    void testGetExceptionMessage_exceptionWithCause() {
-        var ex = new RuntimeException("Root Cause");
-        var exWrapped = new RuntimeException("Wrapped Exception", ex);
-        assertThat(ExceptionUtils.getExceptionMessage(exWrapped))
-                .isEqualTo("Wrapped Exception -> Root Cause");
-    }
-
-    @Test
-    void testGetExceptionMessage_exactlyThree() {
-        var ex3 = new IllegalAccessException(null);
-        var ex2 = new RuntimeException("Cause 2", ex3);
-        var ex = new RuntimeException("Cause 1", ex2);
-        assertThat(ExceptionUtils.getExceptionMessage(ex))
-                .isEqualTo("Cause 1 -> Cause 2 -> IllegalAccessException");
-    }
-
-    @Test
-    void testGetExceptionMessage_onlyTwo() {
-        var ex2 = new RuntimeException("Cause 2");
-        var ex = new RuntimeException("Cause 1", ex2);
-        assertThat(ExceptionUtils.getExceptionMessage(ex)).isEqualTo("Cause 1 -> Cause 2");
-    }
-
-    @Test
-    void testGetExceptionMessage_moreThanThree() {
-        var ex4 = new RuntimeException("Cause 4");
-        var ex3 = new RuntimeException("Cause 3", ex4);
-        var ex2 = new IllegalStateException(null, ex3);
-        var ex = new RuntimeException("Cause 1", ex2);
-        assertThat(ExceptionUtils.getExceptionMessage(ex))
-                .isEqualTo("Cause 1 -> IllegalStateException -> Cause 3");
-    }
-
     @Test
     void testGetExceptionMessage_serializedThrowable() {
         var serializedException = new SerializedThrowable(new Exception("Serialized Exception"));
@@ -85,46 +41,25 @@ void testGetExceptionMessage_serializedThrowable() {
     }
 
     @Test
-    void testGetExceptionMessage_serializedThrowableWithDoubleSerializedException() {
-        var firstSerialized = new SerializedThrowable(new IndexOutOfBoundsException("4>3"));
-        var serializedException = new SerializedThrowable(firstSerialized);
-        assertThat(ExceptionUtils.getExceptionMessage(serializedException)).isEqualTo("4>3");
-    }
-
-    @Test
-    void testGetExceptionMessage_serializedThrowableWithRegularException() {
-        var serializedException = new SerializedThrowable(new Exception("Serialized Exception"));
-        var ex = new RuntimeException("Cause 1", serializedException);
-        assertThat(ExceptionUtils.getExceptionMessage(ex))
-                .isEqualTo("Cause 1 -> Serialized Exception");
-    }
-
-    @Test
-    void testGetExceptionMessage_serializedThrowableAndAnotherCause() {
-        var ex = new RuntimeException("Cause 1");
-        var serializedException =
-                new SerializedThrowable(new RuntimeException("Serialized Exception", ex));
-        assertThat(ExceptionUtils.getExceptionMessage(serializedException))
-                .isEqualTo("Serialized Exception -> Cause 1");
-    }
-
-    @Test
-    void testGetExceptionMessage_moreThanThreeWithSerializedAnnRegularOnes() {
+    void testGetExceptionMessage_differentKindsOfExceptions() {
         var ex4 = new RuntimeException("Cause 4");
         var ex3 = new RuntimeException("Cause 3", ex4);
         var ex2 = new RuntimeException("Cause 2", new SerializedThrowable(ex3));
         var ex = new RuntimeException("Cause 1", ex2);
         assertThat(ExceptionUtils.getExceptionMessage(ex))
-                .isEqualTo("Cause 1 -> Cause 2 -> Cause 3");
+                .isEqualTo("Cause 1 -> Cause 2 -> Cause 3 -> Cause 4");
     }
 
     @Test
-    void testGetExceptionMessage_moreThanThreeWithSerializedAndNotWithDifferentOrdering() {
-        var ex4 = new RuntimeException("Cause 4");
-        var ex3 = new RuntimeException("Cause 3", ex4);
-        var ex2 = new RuntimeException("Cause 2", new SerializedThrowable(ex3));
-        var ex = new RuntimeException("Cause 1", new SerializedThrowable(ex2));
-        assertThat(ExceptionUtils.getExceptionMessage(ex))
-                .isEqualTo("Cause 1 -> Cause 2 -> Cause 3");
+    void testSerializedThrowableError() {
+        var serializedException = new SerializedThrowable(new NonSerializableException());
+        assertThat(ExceptionUtils.getExceptionMessage(serializedException))
+                .isEqualTo("Unknown Error (SerializedThrowable)");
+    }
+
+    private static class NonSerializableException extends Exception {
+        private void writeObject(java.io.ObjectOutputStream stream) throws IOException {
+            throw new IOException();
+        }
     }
 }