Could not restore state after upgrade from 1.11 -> 1.13

[Tom Thornton <th...@yelp.com>]

We have an application running on Flink version 1.11.6. After upgrading the
application to 1.13.5 We saw two errors (full stack traces at end). Both
are "Could not restore keyed state backend for" for AggregateWindowOperator
or KeyedCoProcessOperatorWithWatermarkDelay. The root exception for both is:

InvalidClassException: org.apache.flink.table.types.logical.LogicalType;
local class incompatible: stream classdesc serialVersionUID = <id>, local
class serialVersionUID = <id>

Is this expected behavior that state would not be recoverable between the
two versions? Any ideas how to fix this or what may be causing this issue?
The application uses the kafka connector read from a kafka source, converts
to rows, and uses the blink table planner to execute SQL on the rows, and
then outputs those back to kafka.

Thank you for the help in advance.

Full stack traces:

KeyedCoProcessOperaor error:

2022-04-28 11:20:36java.lang.Exception: Exception while creating
StreamOperatorStateContext.	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)	at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)	at
org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)	at
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)	at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)	at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)	at
java.base/java.lang.Thread.run(Thread.java:829)Caused by:
org.apache.flink.util.FlinkException: Could not restore keyed state
backend for KeyedCoProcessOperatorWithWatermarkDelay_4ab8ac54187ea6bb0315e50bad7fe66a_(1/1)
from any of the 1 provided restore options.	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)	...
10 moreCaused by:
org.apache.flink.runtime.state.BackendBuildingException: Caught
unexpected exception.	at
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:394)	at
org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:465)	at
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:345)	at
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:87)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)	...
12 moreCaused by: java.io.InvalidClassException:
org.apache.flink.table.types.logical.LogicalType; local class
incompatible: stream classdesc serialVersionUID =
-7381419642101800907, local class serialVersionUID = 1	at
java.base/java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:689)	at
java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at
java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at
java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at
java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at
java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2195)	at
java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1681)	at
java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:493)	at
java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:451)	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:593)	at
org.apache.flink.table.runtime.typeutils.RowDataSerializer$RowDataSerializerSnapshot.readSnapshot(RowDataSerializer.java:318)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.readVersionedSnapshot(TypeSerializerSnapshot.java:175)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.deserializeV2(TypeSerializerSnapshotSerializationUtil.java:174)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.read(TypeSerializerSnapshotSerializationUtil.java:145)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(TypeSerializerSnapshotSerializationUtil.java:77)	at
org.apache.flink.runtime.state.KeyedBackendSerializationProxy.read(KeyedBackendSerializationProxy.java:152)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.readMetaData(FullSnapshotRestoreOperation.java:194)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.restoreKeyGroupsInStateHandle(FullSnapshotRestoreOperation.java:171)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.access$100(FullSnapshotRestoreOperation.java:113)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:158)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:140)	at
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:102)	at
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:314)	...
18 more


AggregateWindowOperator error:

2022-04-28 11:23:06java.lang.Exception: Exception while creating
StreamOperatorStateContext.	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)	at
org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)	at
org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)	at
org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)	at
org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)	at
org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)	at
org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)	at
java.base/java.lang.Thread.run(Thread.java:829)Caused by:
org.apache.flink.util.FlinkException: Could not restore keyed state
backend for AggregateWindowOperator_dc8c44a81dc493d97d906777fdf45689_(1/1)
from any of the 1 provided restore options.	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)	...
10 moreCaused by:
org.apache.flink.runtime.state.BackendBuildingException: Caught
unexpected exception.	at
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:394)	at
org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:465)	at
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:345)	at
org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:87)	at
org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)	at
org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)	...
12 moreCaused by: java.io.InvalidClassException:
org.apache.flink.table.types.logical.LogicalType; local class
incompatible: stream classdesc serialVersionUID =
-7381419642101800907, local class serialVersionUID = 1	at
java.base/java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:689)	at
java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at
java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at
java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at
java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at
java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2195)	at
java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1681)	at
java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:493)	at
java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:451)	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)	at
org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:593)	at
org.apache.flink.table.runtime.typeutils.RowDataSerializer$RowDataSerializerSnapshot.readSnapshot(RowDataSerializer.java:318)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.readVersionedSnapshot(TypeSerializerSnapshot.java:175)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.deserializeV2(TypeSerializerSnapshotSerializationUtil.java:174)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.read(TypeSerializerSnapshotSerializationUtil.java:145)	at
org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(TypeSerializerSnapshotSerializationUtil.java:77)	at
org.apache.flink.runtime.state.KeyedBackendSerializationProxy.read(KeyedBackendSerializationProxy.java:152)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.readMetaData(FullSnapshotRestoreOperation.java:194)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.restoreKeyGroupsInStateHandle(FullSnapshotRestoreOperation.java:171)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.access$100(FullSnapshotRestoreOperation.java:113)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:158)	at
org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:140)	at
org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:102)	at
org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:314)	...
18 more

Re: Could not restore state after upgrade from 1.11 -> 1.13

[Guowei Ma <gu...@gmail.com>]

Hi Tom
It seems that you used the Table/Sql api.  Flink cannot guarantee that the
old state could be used by the new version Flink. You could find more
information from the doc. [1]
But there is some work for resolving these situations. You could see the
more information from here [2];

[1]
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/concepts/overview/#stateful-upgrades-and-evolution
[2]
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=191336489

Best,
Guowei


On Sat, Apr 30, 2022 at 1:28 AM Tom Thornton <th...@yelp.com> wrote:

> We have an application running on Flink version 1.11.6. After upgrading
> the application to 1.13.5 We saw two errors (full stack traces at end).
> Both are "Could not restore keyed state backend for" for AggregateWindowOperator
> or KeyedCoProcessOperatorWithWatermarkDelay. The root exception for both
> is:
>
> InvalidClassException: org.apache.flink.table.types.logical.LogicalType;
> local class incompatible: stream classdesc serialVersionUID = <id>, local
> class serialVersionUID = <id>
>
> Is this expected behavior that state would not be recoverable between the
> two versions? Any ideas how to fix this or what may be causing this issue?
> The application uses the kafka connector read from a kafka source, converts
> to rows, and uses the blink table planner to execute SQL on the rows, and
> then outputs those back to kafka.
>
> Thank you for the help in advance.
>
> Full stack traces:
>
> KeyedCoProcessOperaor error:
>
> 2022-04-28 11:20:36java.lang.Exception: Exception while creating StreamOperatorStateContext.	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)	at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)	at java.base/java.lang.Thread.run(Thread.java:829)Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for KeyedCoProcessOperatorWithWatermarkDelay_4ab8ac54187ea6bb0315e50bad7fe66a_(1/1) from any of the 1 provided restore options.	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)	... 10 moreCaused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception.	at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:394)	at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:465)	at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:345)	at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:87)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)	... 12 moreCaused by: java.io.InvalidClassException: org.apache.flink.table.types.logical.LogicalType; local class incompatible: stream classdesc serialVersionUID = -7381419642101800907, local class serialVersionUID = 1	at java.base/java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:689)	at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2195)	at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1681)	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:493)	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:451)	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:593)	at org.apache.flink.table.runtime.typeutils.RowDataSerializer$RowDataSerializerSnapshot.readSnapshot(RowDataSerializer.java:318)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.readVersionedSnapshot(TypeSerializerSnapshot.java:175)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.deserializeV2(TypeSerializerSnapshotSerializationUtil.java:174)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.read(TypeSerializerSnapshotSerializationUtil.java:145)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(TypeSerializerSnapshotSerializationUtil.java:77)	at org.apache.flink.runtime.state.KeyedBackendSerializationProxy.read(KeyedBackendSerializationProxy.java:152)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.readMetaData(FullSnapshotRestoreOperation.java:194)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.restoreKeyGroupsInStateHandle(FullSnapshotRestoreOperation.java:171)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.access$100(FullSnapshotRestoreOperation.java:113)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:158)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:140)	at org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:102)	at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:314)	... 18 more
>
>
> AggregateWindowOperator error:
>
> 2022-04-28 11:23:06java.lang.Exception: Exception while creating StreamOperatorStateContext.	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254)	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272)	at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:441)	at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:585)	at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55)	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeRestore(StreamTask.java:565)	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)	at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:540)	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:759)	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)	at java.base/java.lang.Thread.run(Thread.java:829)Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for AggregateWindowOperator_dc8c44a81dc493d97d906777fdf45689_(1/1) from any of the 1 provided restore options.	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163)	... 10 moreCaused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception.	at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:394)	at org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend.createKeyedStateBackend(EmbeddedRocksDBStateBackend.java:465)	at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:345)	at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:87)	at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328)	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168)	at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135)	... 12 moreCaused by: java.io.InvalidClassException: org.apache.flink.table.types.logical.LogicalType; local class incompatible: stream classdesc serialVersionUID = -7381419642101800907, local class serialVersionUID = 1	at java.base/java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:689)	at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at java.base/java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:2014)	at java.base/java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1864)	at java.base/java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2195)	at java.base/java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1681)	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:493)	at java.base/java.io.ObjectInputStream.readObject(ObjectInputStream.java:451)	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:615)	at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:593)	at org.apache.flink.table.runtime.typeutils.RowDataSerializer$RowDataSerializerSnapshot.readSnapshot(RowDataSerializer.java:318)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshot.readVersionedSnapshot(TypeSerializerSnapshot.java:175)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.deserializeV2(TypeSerializerSnapshotSerializationUtil.java:174)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil$TypeSerializerSnapshotSerializationProxy.read(TypeSerializerSnapshotSerializationUtil.java:145)	at org.apache.flink.api.common.typeutils.TypeSerializerSnapshotSerializationUtil.readSerializerSnapshot(TypeSerializerSnapshotSerializationUtil.java:77)	at org.apache.flink.runtime.state.KeyedBackendSerializationProxy.read(KeyedBackendSerializationProxy.java:152)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.readMetaData(FullSnapshotRestoreOperation.java:194)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.restoreKeyGroupsInStateHandle(FullSnapshotRestoreOperation.java:171)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation.access$100(FullSnapshotRestoreOperation.java:113)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:158)	at org.apache.flink.runtime.state.restore.FullSnapshotRestoreOperation$1.next(FullSnapshotRestoreOperation.java:140)	at org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:102)	at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:314)	... 18 more
>
>