This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 88f68e47f [FLINK-41242] Report the original exception class in
`status.error` (#1233)
88f68e47f is described below
commit 88f68e47f2b6249aa38c7e5f0decf3cd3bdc5407
Author: Santwana Verma <[email protected]>
AuthorDate: Thu Oct 29 12:52:27 2026 +0530
[FLINK-41242] Report the original exception class in `status.error` (#1233)
* [FLINK-41242] Report the original exception class in status.error
The JobManager returns a failed job's cause as a SerializedThrowable, and
the operator took the type in status.error from the class of the throwable
it holds, so every type read org.apache.flink.util.SerializedThrowable.
Use the original class name that SerializedThrowable records instead. This
applies to any SerializedThrowable in the reported error and its causes,
most often a failed job's failure cause.
Generated-by: Claude Code (Claude Opus 5.5)
* [FLINK-41242] Move the SerializedThrowable class name lookup to
ExceptionUtils
Generated-by: Claude Code (Claude Opus 5.5)
* [FLINK-41242] Use a static import for ExceptionUtils#getExceptionClassName
Generated-by: Claude Code (Claude Opus 5.5)
---------
Co-authored-by: Santwana Verma <[email protected]>
---
.../kubernetes/operator/utils/ExceptionUtils.java | 13 ++++++++++
.../utils/FlinkResourceExceptionUtils.java | 4 ++-
.../deployment/ApplicationObserverTest.java | 10 ++++++++
.../operator/utils/ExceptionUtilsTest.java | 14 ++++++++++
.../utils/FlinkResourceExceptionUtilsTest.java | 30 ++++++++++++++++++++++
5 files changed, 70 insertions(+), 1 deletion(-)
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtils.java
index 002c43b79..f47db93c9 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtils.java
@@ -66,6 +66,19 @@ public class ExceptionUtils {
return Optional.empty();
}
+ /**
+ * Returns the class name of the throwable. For a {@link
SerializedThrowable}, returns the class
+ * name of the original error it wraps.
+ *
+ * @param throwable the throwable to be processed
+ * @return the class name of the throwable, or of the original error for a
SerializedThrowable
+ */
+ public static String getExceptionClassName(Throwable throwable) {
+ return throwable instanceof SerializedThrowable
+ ? ((SerializedThrowable) throwable).getOriginalErrorClassName()
+ : throwable.getClass().getName();
+ }
+
/**
* traverse the throwable and extract useful information for up to the
first 3 possible
* exceptions in the hierarchy.
diff --git
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtils.java
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtils.java
index 9647ac2ae..97aaff9b0 100644
---
a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtils.java
+++
b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtils.java
@@ -37,6 +37,8 @@ import java.util.Optional;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
+import static
org.apache.flink.kubernetes.operator.utils.ExceptionUtils.getExceptionClassName;
+
/** Flink Resource Exception utilities. */
public final class FlinkResourceExceptionUtils {
@@ -122,7 +124,7 @@ public final class FlinkResourceExceptionUtils {
Map<String, String> labelMapper) {
FlinkResourceException flinkResourceException =
FlinkResourceException.builder().build();
- getSubstringWithMaxLength(throwable.getClass().getName(),
lengthThreshold)
+ getSubstringWithMaxLength(getExceptionClassName(throwable),
lengthThreshold)
.ifPresent(flinkResourceException::setType);
getSubstringWithMaxLength(throwable.getMessage(), lengthThreshold)
.ifPresent(flinkResourceException::setMessage);
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
index 67265eaf7..433d01b96 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/deployment/ApplicationObserverTest.java
@@ -35,6 +35,7 @@ import
org.apache.flink.kubernetes.operator.api.status.TaskManagerInfo;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import
org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.exception.FlinkResourceException;
import org.apache.flink.kubernetes.operator.observer.TestObserverAdapter;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
@@ -42,6 +43,7 @@ import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
+import com.fasterxml.jackson.databind.ObjectMapper;
import io.fabric8.kubernetes.api.model.Event;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
@@ -62,6 +64,7 @@ import java.util.stream.Stream;
import static
org.apache.flink.kubernetes.operator.api.utils.FlinkResourceUtils.getCheckpointInfo;
import static
org.apache.flink.kubernetes.operator.api.utils.FlinkResourceUtils.getJobStatus;
import static
org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
@@ -270,6 +273,13 @@ public class ApplicationObserverTest extends
OperatorTestBase {
observer.observe(deployment, readyContext);
Assertions.assertEquals(0, flinkService.getRunningCount());
Assertions.assertTrue(deployment.getStatus().getError().contains("Job
failed"));
+ assertThat(
+ new ObjectMapper()
+ .readValue(
+ deployment.getStatus().getError(),
+ FlinkResourceException.class)
+ .getType())
+ .isEqualTo(RuntimeException.class.getName());
}
@Test
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtilsTest.java
index 144ec3189..c48073060 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtilsTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/ExceptionUtilsTest.java
@@ -63,6 +63,20 @@ public class ExceptionUtilsTest {
.isEqualTo(NonSerializableException.class.getName());
}
+ @Test
+ void testGetExceptionClassName() {
+ assertThat(ExceptionUtils.getExceptionClassName(new
IllegalStateException()))
+ .isEqualTo(IllegalStateException.class.getName());
+ assertThat(
+ ExceptionUtils.getExceptionClassName(
+ new SerializedThrowable(new
IllegalStateException())))
+ .isEqualTo(IllegalStateException.class.getName());
+ assertThat(
+ ExceptionUtils.getExceptionClassName(
+ new SerializedThrowable(new
NonSerializableException())))
+ .isEqualTo(NonSerializableException.class.getName());
+ }
+
private static class NonSerializableException extends Exception {
public NonSerializableException(String message) {
diff --git
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtilsTest.java
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtilsTest.java
index f66f84b75..2e46b7347 100644
---
a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtilsTest.java
+++
b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/FlinkResourceExceptionUtilsTest.java
@@ -30,6 +30,7 @@ import
org.apache.flink.kubernetes.operator.exception.MissingJobManagerException
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
import org.apache.flink.runtime.rest.util.RestClientException;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.SerializedThrowable;
import
org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
@@ -49,6 +50,7 @@ import java.util.stream.Stream;
import static
org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.LABELS;
import static
org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.getSubstringWithMaxLength;
import static
org.apache.flink.kubernetes.operator.utils.FlinkResourceExceptionUtils.updateFlinkResourceException;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -282,6 +284,34 @@ public class FlinkResourceExceptionUtilsTest {
}
}
+ @Test
+ public void testSerializedThrowableReportsOriginalType() throws
JsonProcessingException {
+ FlinkConfigManager configManager = new FlinkConfigManager(new
Configuration());
+ SerializedThrowable serializedThrowable =
+ new SerializedThrowable(
+ new FlinkRuntimeException(
+ "job failed", new NumberFormatException("For
input string: x1")));
+
+ for (AbstractFlinkResource resource : getTestResources()) {
+ updateFlinkResourceException(
+ serializedThrowable, resource,
configManager.getOperatorConfiguration());
+
+ FlinkResourceException flinkResourceException =
+ new ObjectMapper()
+ .readValue(
+ ((CommonStatus)
resource.getStatus()).getError(),
+ FlinkResourceException.class);
+
+ assertThat(flinkResourceException.getType())
+ .isEqualTo(FlinkRuntimeException.class.getName());
+ assertThat(flinkResourceException.getMessage())
+ .isEqualTo(serializedThrowable.getMessage());
+ assertThat(flinkResourceException.getThrowableList())
+ .extracting(FlinkResourceException::getType)
+ .containsExactly(NumberFormatException.class.getName());
+ }
+ }
+
private static ReconciliationException getTestException() {
return new ReconciliationException(
"reconciliation exception message",
