[Schema Evolution] Cannot restore from savepoint after deleting field from POJO
Locked
 
|
Hi everyone, It seems I’m having either the same problem, or a problem similar to the one mentioned here:
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Problem-when-restoring-from-savepoint-with-missing-state-amp-POJO-modification-td39945.html I have a POJO class that is used in Flink state. The class is annotated with @TypeInfo as described, e.g., here:
https://stackoverflow.com/a/64721838/5793905 Now I want to remove a field from the POJO. This removal is also considered in the corresponding TypeInfoFactory. However, after trying to restore from a savepoint where the POJO still had the field I get this exception: 2021-03-10T20:51:30.406Z INFO org.apache.flink.runtime.taskmanager.Task:960 … (6/8) (d630d5ff0d7ae4fbc428b151abebab52) switched from RUNNING to FAILED. java.lang.Exception: Exception while creating StreamOperatorStateContext. at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195) at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:253) at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:901) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:415) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) at java.lang.Thread.run(Thread.java:748) Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for KeyedCoProcessOperator_c535ac415eeb524d67c88f4a481077d2_(6/8) from any of the 1 provided restore options. at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:307) at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135) ... 6 common frames omitted Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116) at org.apache.flink.runtime.state.memory.MemoryStateBackend.createKeyedStateBackend(MemoryStateBackend.java:347) at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:291) at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142) at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121) ... 8 common frames omitted Caused by: java.lang.NullPointerException: null at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.<init>(PojoSerializer.java:123) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:186) at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:56) at org.apache.flink.api.common.typeutils.CompositeSerializer$PrecomputedParameters.precompute(CompositeSerializer.java:228) at org.apache.flink.api.common.typeutils.CompositeSerializer.<init>(CompositeSerializer.java:51) at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializer.<init>(TtlStateFactory.java:250) at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:359) at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:330) at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948) at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546) at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260) at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505) at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225) at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83) at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189) at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164) at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136) at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315) at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54) at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36) at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable.java:98) at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(CopyOnWriteStateTable.java:49) at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41) at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243) at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185) at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152) at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114) ... 12 common frames omitted I’m currently using Flink 1.9.3 to both save and restore the job’s savepoint. Even though I can avoid the problem by simply leaving the field in the class, a problem like this would prevent rollback, which is important for my application. Regards, Alexis. |
Re: [Schema Evolution] Cannot restore from savepoint after deleting field from POJO
|
|
Hi Alexis,
This looks like a bug, I've created a Jira ticket to address it [1]. Please feel free to provide any additional information. In particular, whether you are able to reproduce it in any of the subsequent releases. [1] https://issues.apache.org/jira/browse/FLINK-21752 Regards, Roman On Thu, Mar 11, 2021 at 5:36 PM Alexis Sarda-Espinosa <[hidden email]> wrote: > > Hi everyone, > > > > It seems I’m having either the same problem, or a problem similar to the one mentioned here: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Problem-when-restoring-from-savepoint-with-missing-state-amp-POJO-modification-td39945.html > > > > I have a POJO class that is used in Flink state. The class is annotated with @TypeInfo as described, e.g., here: https://stackoverflow.com/a/64721838/5793905 > > > > Now I want to remove a field from the POJO. This removal is also considered in the corresponding TypeInfoFactory. However, after trying to restore from a savepoint where the POJO still had the field I get this exception: > > > > 2021-03-10T20:51:30.406Z INFO org.apache.flink.runtime.taskmanager.Task:960 … (6/8) (d630d5ff0d7ae4fbc428b151abebab52) switched from RUNNING to FAILED. java.lang.Exception: Exception while creating StreamOperatorStateContext. > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195) > > at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:253) > > at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:901) > > at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:415) > > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) > > at java.lang.Thread.run(Thread.java:748) > > Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for KeyedCoProcessOperator_c535ac415eeb524d67c88f4a481077d2_(6/8) from any of the 1 provided restore options. > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:307) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135) > > ... 6 common frames omitted > > Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend > > at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116) > > at org.apache.flink.runtime.state.memory.MemoryStateBackend.createKeyedStateBackend(MemoryStateBackend.java:347) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:291) > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142) > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121) > > ... 8 common frames omitted > > Caused by: java.lang.NullPointerException: null > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.<init>(PojoSerializer.java:123) > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:186) > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:56) > > at org.apache.flink.api.common.typeutils.CompositeSerializer$PrecomputedParameters.precompute(CompositeSerializer.java:228) > > at org.apache.flink.api.common.typeutils.CompositeSerializer.<init>(CompositeSerializer.java:51) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializer.<init>(TtlStateFactory.java:250) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:359) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:330) > > at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) > > at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) > > at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948) > > at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) > > at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) > > at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546) > > at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260) > > at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505) > > at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225) > > at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83) > > at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) > > at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189) > > at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164) > > at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136) > > at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36) > > at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable.java:98) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(CopyOnWriteStateTable.java:49) > > at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41) > > at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243) > > at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185) > > at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152) > > at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114) > > ... 12 common frames omitted > > > > I’m currently using Flink 1.9.3 to both save and restore the job’s savepoint. Even though I can avoid the problem by simply leaving the field in the class, a problem like this would prevent rollback, which is important for my application. > > > > Regards, > > Alexis. > > |
|
|
Hello,
I see that new Jira bots are now active. If no one has time to look at this, could documentation at least be updated to reflect the fact that removing fields from POJOs will break state restoration?
Regards,
Alexis.
From: Roman Khachatryan <[hidden email]>
Sent: Friday, March 12, 2021 6:22 PM To: Alexis Sarda-Espinosa <[hidden email]> Cc: [hidden email] <[hidden email]> Subject: Re: [Schema Evolution] Cannot restore from savepoint after deleting field from POJO Hi Alexis,
This looks like a bug, I've created a Jira ticket to address it [1]. Please feel free to provide any additional information. In particular, whether you are able to reproduce it in any of the subsequent releases. [1] https://issues.apache.org/jira/browse/FLINK-21752 Regards, Roman On Thu, Mar 11, 2021 at 5:36 PM Alexis Sarda-Espinosa <[hidden email]> wrote: > > Hi everyone, > > > > It seems I’m having either the same problem, or a problem similar to the one mentioned here: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Problem-when-restoring-from-savepoint-with-missing-state-amp-POJO-modification-td39945.html > > > > I have a POJO class that is used in Flink state. The class is annotated with @TypeInfo as described, e.g., here: https://stackoverflow.com/a/64721838/5793905 > > > > Now I want to remove a field from the POJO. This removal is also considered in the corresponding TypeInfoFactory. However, after trying to restore from a savepoint where the POJO still had the field I get this exception: > > > > 2021-03-10T20:51:30.406Z INFO org.apache.flink.runtime.taskmanager.Task:960 … (6/8) (d630d5ff0d7ae4fbc428b151abebab52) switched from RUNNING to FAILED. java.lang.Exception: Exception while creating StreamOperatorStateContext. > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:195) > > at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:253) > > at org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:901) > > at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:415) > > at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) > > at org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) > > at java.lang.Thread.run(Thread.java:748) > > Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for KeyedCoProcessOperator_c535ac415eeb524d67c88f4a481077d2_(6/8) from any of the 1 provided restore options. > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:307) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:135) > > ... 6 common frames omitted > > Caused by: org.apache.flink.runtime.state.BackendBuildingException: Failed when trying to restore heap backend > > at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:116) > > at org.apache.flink.runtime.state.memory.MemoryStateBackend.createKeyedStateBackend(MemoryStateBackend.java:347) > > at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:291) > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:142) > > at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:121) > > ... 8 common frames omitted > > Caused by: java.lang.NullPointerException: null > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.<init>(PojoSerializer.java:123) > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:186) > > at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:56) > > at org.apache.flink.api.common.typeutils.CompositeSerializer$PrecomputedParameters.precompute(CompositeSerializer.java:228) > > at org.apache.flink.api.common.typeutils.CompositeSerializer.<init>(CompositeSerializer.java:51) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializer.<init>(TtlStateFactory.java:250) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:359) > > at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:330) > > at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) > > at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193) > > at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948) > > at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482) > > at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472) > > at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:546) > > at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260) > > at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:505) > > at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.snapshotsToRestoreSerializers(NestedSerializersSnapshotDelegate.java:225) > > at org.apache.flink.api.common.typeutils.NestedSerializersSnapshotDelegate.getRestoredNestedSerializers(NestedSerializersSnapshotDelegate.java:83) > > at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:194) > > at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:189) > > at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:164) > > at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:136) > > at org.apache.flink.runtime.state.heap.StateTable.getStateSerializer(StateTable.java:315) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:54) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.createStateMap(CopyOnWriteStateTable.java:36) > > at org.apache.flink.runtime.state.heap.StateTable.<init>(StateTable.java:98) > > at org.apache.flink.runtime.state.heap.CopyOnWriteStateTable.<init>(CopyOnWriteStateTable.java:49) > > at org.apache.flink.runtime.state.heap.AsyncSnapshotStrategySynchronicityBehavior.newStateTable(AsyncSnapshotStrategySynchronicityBehavior.java:41) > > at org.apache.flink.runtime.state.heap.HeapSnapshotStrategy.newStateTable(HeapSnapshotStrategy.java:243) > > at org.apache.flink.runtime.state.heap.HeapRestoreOperation.createOrCheckStateForMetaInfo(HeapRestoreOperation.java:185) > > at org.apache.flink.runtime.state.heap.HeapRestoreOperation.restore(HeapRestoreOperation.java:152) > > at org.apache.flink.runtime.state.heap.HeapKeyedStateBackendBuilder.build(HeapKeyedStateBackendBuilder.java:114) > > ... 12 common frames omitted > > > > I’m currently using Flink 1.9.3 to both save and restore the job’s savepoint. Even though I can avoid the problem by simply leaving the field in the class, a problem like this would prevent rollback, which is important for my application. > > > > Regards, > > Alexis. > > |
| Free forum by Nabble | Edit this page |