Flink 的 QueryableStateClient 抛出“FlinkRuntimeException: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 85”

如何解决Flink 的 QueryableStateClient 抛出“FlinkRuntimeException: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 85”

我正在尝试使用其 QueryableStateClient API 在 Flink 作业中持久化的自定义对象上查询 ValueState。我可以观察到持久性工作正常，但是，来自客户端应用程序的查询抛出 FlinkRuntimeException。

以下是客户端应用程序中使用的代码片段：

ValueStateDescriptor<CustomObj> descriptor =new ValueStateDescriptor<>("StateDesc",Typeinformation.of(new TypeHint<>() {}).createSerializer(new ExecutionConfig()));

QueryableStateClient client = new QueryableStateClient(proxyHostName,proxyPort);
CompletableFuture<ValueState<CustomObj>> res =client.getKvstate(jobId,"state-query",key,BasicTypeInfo.STRING_TYPE_INFO,descriptor);

当我执行 res.get() 时，以下是我得到的错误：

Caused by: java.util.concurrent.ExecutionException: org.apache.flink.util.FlinkRuntimeException: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 85
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
    ... 134 common frames omitted
Caused by: org.apache.flink.util.FlinkRuntimeException: com.esotericsoftware.kryo.KryoException: Encountered unregistered class ID: 85
    at org.apache.flink.queryablestate.client.QueryableStateClient.createState(QueryableStateClient.java:289)
    at org.apache.flink.queryablestate.client.QueryableStateClient.lambda$getKvstate$2(QueryableStateClient.java:274)
    at java.base/java.util.concurrent.CompletableFuture$uniapply.tryFire(CompletableFuture.java:642)
    at java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
    at java.base/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
    at org.apache.flink.queryablestate.network.Client$PendingConnection.lambda$handInChannel$0(Client.java:326)

flink 作业中用于持久化状态的代码片段：

ValueStateDescriptor<CustomObj> descriptor =new ValueStateDescriptor<>("StateDesc",Typeinformation.of(new TypeHint<>() {}));
descriptor.setQueryable("state-query");
state = getRuntimeContext().getState(descriptor);

我也尝试过 Types.GENERIC(CustomObj.class) 使用描述符，但没有用。仅供参考，我在我的 flink 作业和客户端应用程序中都使用了我的 CustomObj 的精确副本。如果我做错了什么，请强调。