flink rest api client 解析服务端返回的结果有类无法反序列化

[baisui1981]

Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.concurrent.FutureUtils$RetryException: Could not complete the operation. Number of retries has been exhausted.
    at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
    at com.qlangtech.tis.plugins.flink.client.FlinkClient.lambda$null$1(FlinkClient.java:103)
    at org.apache.flink.client.ClientUtils.waitUntilJobInitializationFinished(ClientUtils.java:164)
    ... 9 common frames omitted
Caused by: java.io.IOException: Failed to deserialize org.apache.flink.util.SerializedThrowable
    at org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer.deserialize(SerializedThrowableDeserializer.java:53)
    at org.apache.flink.runtime.rest.messages.json.JobResultDeserializer.deserialize(JobResultDeserializer.java:106)
    at org.apache.flink.runtime.rest.messages.json.JobResultDeserializer.deserialize(JobResultDeserializer.java:50)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.SettableBeanProperty.deserialize(SettableBeanProperty.java:542)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeWithErrorWrapping(BeanDeserializer.java:565)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.BeanDeserializer._deserializeUsingPropertyBased(BeanDeserializer.java:449)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.BeanDeserializerBase.deserializeFromObjectUsingNonDefault(BeanDeserializerBase.java:1390)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.BeanDeserializer.deserializeFromObject(BeanDeserializer.java:362)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:195)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:322)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper._readValue(ObjectMapper.java:4569)
    at org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2867)
    at org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:475)
    ... 7 common frames omitted
Caused by: java.lang.ClassNotFoundException: org.apache.flink.util.SerializedThrowable
    at java.net.URLClassLoader.findClass(URLClassLoader.java:387)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:355)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
    at java.lang.Class.forName0(Native Method)
    at java.lang.Class.forName(Class.java:348)
    at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:76)
    at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2011)
    at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1875)
    at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2209)

原因是 org.apache.flink.runtime.rest.messages.json.SerializedThrowableDeserializer 类中无法从classloader中加载flink的类：

public class SerializedThrowableDeserializer extends StdDeserializer<SerializedThrowable> {
    private static final long serialVersionUID = 1L;

    public SerializedThrowableDeserializer() {
        super(SerializedThrowable.class);
    }

    public SerializedThrowable deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
        JsonNode root = (JsonNode)p.readValueAsTree();
        byte[] serializedException = root.get("serialized-throwable").binaryValue();

        try {
            return (SerializedThrowable)InstantiationUtil.deserializeObject(serializedException, ClassLoader.getSystemClassLoader());
        } catch (ClassNotFoundException var6) {
            throw new IOException("Failed to deserialize " + SerializedThrowable.class.getCanonicalName(), var6);
        }
    }
}