[FLINK-40674][runtime] Stop deserializing SerializedThrowable via ObjectInputStream on parse

[gaborgsomogyi]

What is the purpose of the change

SerializedThrowableDeserializer currently reconstructs the serialized-throwable REST field by running it through ObjectInputStream.readObject() as an automatic side effect of parsing, even though none of the current callers of this field actually need the live, reconstructed exception object at that point; they only ever read its class name, message, and stack trace as text.

This pull request changes SerializedThrowableSerializer/SerializedThrowableDeserializer to carry that information as additive, plain-text JSON fields, so parsing a response recovers it directly, without touching Java's own object deserialization at all. The original exception object remains available through the existing SerializedThrowable#deserializeError() method, via an explicit, separate call, for the handful of callers that still need it.

It changes the REST API in a backward compatible way.
Before:

{
  "class": "java.lang.RuntimeException",
  "stack-trace": "java.lang.RuntimeException: boom\n\tat ...",
  "serialized-throwable": "<base64 Java-serialized SerializedThrowable, unbounded>"
}

After:

{
  "class": "java.lang.RuntimeException",
  "message": "java.lang.RuntimeException: boom",
  "stack-trace": "java.lang.RuntimeException: boom\n\tat ...",
  "serialized-throwable": "<base64 blob, kept only for old readers>",
  "cause": {
    "class": "java.io.IOException",
    "message": "java.io.IOException: root cause",
    "stack-trace": "...",
    "serialized-throwable": "..."
  },
  "suppressed": [
    { "class": "...", "message": "...", "stack-trace": "...", "serialized-throwable": "..." }
  ]
}

Brief change log

• SerializedThrowableSerializer now also writes message, cause, and suppressed (recursively) as plain JSON fields, alongside the existing class, stack-trace, and serialized-throwable fields, which keep their current name, meaning, and byte content unchanged.
• SerializedThrowableDeserializer reconstructs a SerializedThrowable, and its cause/suppressed chain, entirely from those text fields, with a bounded recursion depth, and no longer calls ObjectInputStream.readObject() while parsing.
• SerializedThrowable gains a constructor that builds an instance directly from the text fields above, and its structured getStackTrace() no longer reflects the deserializer's own call stack.
• JobResult gains toSafeJobExecutionResult(ClassLoader), which builds the failure exception without deserializing its cause. ClusterClientJobClientAdapter, which backs any RestClusterClient-based JobClient (including remote/session-cluster job submission), now uses it. The existing toJobExecutionResult(ClassLoader) keeps its original behavior unchanged for same-process callers (MiniCluster, Application Mode).
• ClientUtils#waitUntilJobInitializationFinished reconstructs JobInitializationException from the text fields above instead of calling deserializeError().

Verifying this change

This change added tests and can be verified as follows:

• SerializedThrowableSerializerTest: round-trip tests confirming class/message/stack-trace/cause/suppressed survive parsing without deserialization; a test with a non-Java-serialized byte payload in serialized-throwable confirming parsing never attempts deserialization regardless of what that field contains; two tests covering both rolling-upgrade directions (an old deserializer's logic against this fix's serializer output, and this fix's deserializer against an old-shaped response); a test confirming the reconstructed exception's structured stack trace does not leak the deserializer's own call stack; a test confirming a cause chain nested past a fixed depth fails parsing with an IOException instead of exhausting the stack.
• RestClusterClientJobResultSafetyTest drives the real RestClusterClient.requestJobResult() against a fake server returning a response whose serialized-throwable field is set so that a marker flips if that field is ever read via ObjectInputStream.readObject(). Confirms the marker never fires while parsing, and only fires once deserializeError() is called explicitly afterward.
• JobResultTest covers toSafeJobExecutionResult()'s behavior (cause stays a SerializedThrowable) alongside the unchanged toJobExecutionResult() (cause is deserialized, as before).

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: yes (SerializedThrowableSerializer/SerializedThrowableDeserializer gain additive wire fields; both rolling-upgrade directions are covered by tests)
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no (this changes how a job's failure cause is reported and parsed, not the deployment or recovery logic itself)
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no
• If yes, how is the feature documented? not applicable

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Claude code

[flinkbot]

CI report:

• b68b71f Azure: FAILURE
• d03c0ad Azure: PENDING

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build