flink sql 从savepoint 重启遭遇新旧状态序列化不匹配的问题

[mangguozhi <ro...@gmail.com>]

各位好，我在flink 1.13中使用flink sql 在一次修改代码后的重启任务中，报以下错误：

For heap backends, the new state serializer
(org.apache.flink.api.common.typeutils.base.MapSerializer@a5b17bdb) must not
be incompatible with the old state serializer
(org.apache.flink.api.common.typeutils.base.MapSerializer@e5a9c6d8).


我更改了sql中的一个时间字段

CURRENT_TIMESTAMP --> 事件时间的字段 ts  

ts 属性由下面转换而来：

ts AS TO_TIMESTAMP(FROM_UNIXTIME(tts, 'yyyy-MM-dd HH:mm:ss'))

代码层面改动如下：
checkpoint 保存时间调大：

 env.getCheckpointConfig().setCheckpointInterval(600000L);

允许保留比原来更多的checkpoint个数：  
 
 env.getCheckpointConfig().setMaxConcurrentCheckpoints(20);

新增空闲状态过期时间：

Configuration configuration = tableEnv.getConfig().getConfiguration();
tableEnv.getConfig().setIdleStateRetention(Duration.ofDays(1));

想请问一下，这种改动为什么会导致新旧状态不匹配，又该如何避免呢？



--
Sent from: http://apache-flink.147419.n8.nabble.com/