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 <alexis.sarda-espin...@microfocus.com> 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. > >
