serialized throwable fixup

Author: gyfora

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

@@ -93,11 +93,12 @@ private static String getExceptionMessage(Throwable throwable, int level) {
         }
 
         if (throwable instanceof SerializedThrowable) {
+            var serialized = ((SerializedThrowable) throwable);
             var deserialized =
-                    ((SerializedThrowable) throwable)
-                            .deserializeError(Thread.currentThread().getContextClassLoader());
+                    serialized.deserializeError(Thread.currentThread().getContextClassLoader());
             if (deserialized == throwable) {
-                return "Unknown Error (SerializedThrowable)";
+                var msg = serialized.getMessage();
+                return msg != null ? msg : serialized.getOriginalErrorClassName();
             } else {
                 return getExceptionMessage(deserialized, level);
             }

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

@@ -52,12 +52,25 @@ void testGetExceptionMessage_differentKindsOfExceptions() {
 
     @Test
     void testSerializedThrowableError() {
-        var serializedException = new SerializedThrowable(new NonSerializableException());
-        assertThat(ExceptionUtils.getExceptionMessage(serializedException))
-                .isEqualTo("Unknown Error (SerializedThrowable)");
+        assertThat(
+                        ExceptionUtils.getExceptionMessage(
+                                new SerializedThrowable(new NonSerializableException("Message"))))
+                .isEqualTo(String.format("%s: Message", NonSerializableException.class.getName()));
+
+        assertThat(
+                        ExceptionUtils.getExceptionMessage(
+                                new SerializedThrowable(new NonSerializableException())))
+                .isEqualTo(NonSerializableException.class.getName());
     }
 
     private static class NonSerializableException extends Exception {
+
+        public NonSerializableException(String message) {
+            super(message);
+        }
+
+        public NonSerializableException() {}
+
         private void writeObject(java.io.ObjectOutputStream stream) throws IOException {
             throw new IOException();
         }