Flink-新增 Kafka source 引发状态丢失导致启动失败

背景

Flink Job 新增 kafka source 算子，从状态保留并启动后提示 org.apache.flink.util.StateMigrationException: The new state typeSerializer for operator state must not be incompatible，导致任务 Fail。

Source: task-kafka-source -> task-kafka-transform (1/2)#0 (84abd95cd07520de0f292ffd4cb6c01b) switched from RUNNING to FAILED.
org.apache.flink.util.StateMigrationException: The new state typeSerializer for operator state must not be incompatible.at org.apache.flink.runtime.state.DefaultOperatorStateBackend.getListState(DefaultOperatorStateBackend.java:298)at org.apache.flink.runtime.state.DefaultOperatorStateBackend.getUnionListState(DefaultOperatorStateBackend.java:219)at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.initializeState(FlinkKafkaConsumerBase.java:920)at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:185)at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:167)at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)at org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.initializeOperatorState(StreamOperatorStateHandler.java:107)at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:264)at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:400)at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$2(StreamTask.java:507)at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor