社区各位大佬，我遇到了一个Flink版本升级后，savepoint不会正确恢复问题。
版本从1.9.3 升级到1.11.0或1.11.3.
连接器使用的pulsar的连接器。在Source的checkpoint数据结构如下
oldUnionOffsetStates = stateStore.getUnionListState(
new ListStateDescriptor<>(
OFFSETS_STATE_NAME,
TypeInformation.of(new TypeHint>() {
新的集群使用的是更新之后的 pulsar connector ？我看了下 pulsar-flink 的代码，这个更新对 state 是破坏性的。
+unionOffsetStates = stateStore.getUnionListState(
+new ListStateDescriptor<>(
+OFFSETS_STATE_NAME,
+TypeInformation.of(new TypeHint>() {
+
现在是我在维护pulsar-flink
connector，是存在不兼容的升级。还是个很坑的改动。我现在尝试旧的迁移新的字段上方法，会报这个错误。我对1.11支持的代码进行修改，将state的数据结构改成旧版本的形式，同样也是这个错误。你说的StatefulSinkWriterOperator我研究下怎么使用。
2021年3月11日 上午11:36，Kezhu Wang mailto:kez...@gmail.com>> 写道：
StatefulSinkWriterOperator
你好，我参考StatefulSinkWriterOperator重写了迁移方法，进行测试，还是遇到了上面的错误。
这个错误似乎发生在source的initializeState之前。有没有可能flink在读取savepoint文件时，就做了反序列化，在这个过程中失败了。请问有什么办法能帮忙定位吗？
感谢~
2021年3月11日 上午11:36，Kezhu Wang mailto:kez...@gmail.com>> 写道：
新的集群使用的是更新之后的 pulsar connector ？我看了下 pulsar-flink 的代码，这个更新对 state 是破坏性的。
+
> 有没有可能flink在读取savepoint文件时，就做了反序列化，在这个过程中失败了。
确实是这样的，checkpoint 把 serializer 也 snapshot 了。
重新看了下 stack，应该是 deserialize `MessageId` 的时候出错的。你可以看下，pulsar
的版本是不是也有变动？有的话，这两个版本之间的某个 `MessageId` 实现是不是有字段变动？感觉你们应该用
`MessageId.toByteArray`。
On March 11, 2021 at 20:26:15, 赵 建云 (zhaojianyu...@outlook.co
确认了，pulsar的MessageId的实现类内部的增加了字段，导致flink在反序列化时失败了。具体的issue：https://github.com/streamnative/pulsar-flink/issues/256。
我会给flink 1.9的pulsar连接器升级下checkpoint，让MessageId的序列化使用基于
`MessageId.toByteArray`的序列化器。
非常感谢您的帮助~。
Jianyun8023
2021-03-12
2021年3月11日 下午10:43，Kezhu Wang mailto:kez...@gmail.com>> 写道：
建云，
之前我也遇到了savepoint 起作业失败的问题，是我们升级pulsar客户端以后，从2.2升级到2.5.2，我-s
启动作业的时候。因为作业也不是很重要，当时手头有其他任务，我就没有关注这个问题。你看看pulsar source那儿是不是做了什么。
| |
allanqinjy
|
|
allanqi...@163.com
|
签名由网易邮箱大师定制
在2021年03月11日 22:43，Kezhu Wang 写道：
有没有可能flink在读取savepoint文件时，就做了反序列化，在这个过程中失败了。
确实是这样的，checkpoint 把 seriali
