[FLINK-22761] Cannot remove POJO fields

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.12.1
• Component/s: API / Type Serialization System
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Ygor Allan de Fraga
• Created: 24/May/21 15:02
• Updated: 26/Apr/22 23:19

Description

I tested a schema evolution in a state using POJO and no problem was found when trying to add a new field, it was executed just fine. This same field was removed from the POJO as it was just a test, but the application could not restore the state due to an error.

 

Here is the error:

2021-05-24 13:05:31,958 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Co-Flat Map -> Map (3/3)#464 (e0e6d41a18214eab0a1d3c089d8672de) switched from RUNNING to FAILED.2021-05-24 13:05:31,958 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Co-Flat Map -> Map (3/3)#464 (e0e6d41a18214eab0a1d3c089d8672de) switched from RUNNING to FAILED.java.lang.Exception: Exception while creating StreamOperatorStateContext. at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:254) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:272) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.runtime.tasks.OperatorChain.initializeStateAndOpenOperators(OperatorChain.java:425) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$beforeInvoke$2(StreamTask.java:535) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.beforeInvoke(StreamTask.java:525) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:565) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755) [zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570) [zdata-flink-streams.jar:0.1] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_282]

Caused by: org.apache.flink.util.FlinkException: Could not restore keyed state backend for CoStreamFlatMap_b101f370952ea85c2104e98dd54bf7f9_(3/3) from any of the 1 provided restore options. at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:160) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[zdata-flink-streams.jar:0.1] ... 9 more

Caused by: org.apache.flink.runtime.state.BackendBuildingException: Caught unexpected exception. at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:361) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:587) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:93) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[zdata-flink-streams.jar:0.1] ... 9 more

Caused by: java.lang.NullPointerException at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.<init>(PojoSerializer.java:119) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:184) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:56) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:170) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.java.typeutils.runtime.PojoSerializer.duplicate(PojoSerializer.java:56) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.common.typeutils.CompositeSerializer$PrecomputedParameters.precompute(CompositeSerializer.java:239) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.common.typeutils.CompositeSerializer.<init>(CompositeSerializer.java:52) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializer.<init>(TtlStateFactory.java:270) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:387) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.ttl.TtlStateFactory$TtlSerializerSnapshot.createOuterSerializerWithNestedSerializers(TtlStateFactory.java:354) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.api.common.typeutils.CompositeTypeSerializerSnapshot.restoreSerializer(CompositeTypeSerializerSnapshot.java:218) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.StateSerializerProvider.previousSchemaSerializer(StateSerializerProvider.java:186) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.StateSerializerProvider.currentSchemaSerializer(StateSerializerProvider.java:161) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.runtime.state.RegisteredKeyValueStateBackendMetaInfo.getStateSerializer(RegisteredKeyValueStateBackendMetaInfo.java:143) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.ttl.RocksDbTtlCompactFiltersManager.setAndRegisterCompactFilterIfStateTtl(RocksDbTtlCompactFiltersManager.java:73) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.createColumnFamilyDescriptor(RocksDBOperationUtils.java:167) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBOperationUtils.createStateInfo(RocksDBOperationUtils.java:144) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.restore.AbstractRocksDBRestoreOperation.getOrRegisterStateColumnFamilyHandle(AbstractRocksDBRestoreOperation.java:178) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKVStateMetaData(RocksDBFullRestoreOperation.java:186) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restoreKeyGroupsInStateHandle(RocksDBFullRestoreOperation.java:157) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.restore.RocksDBFullRestoreOperation.restore(RocksDBFullRestoreOperation.java:142) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackendBuilder.build(RocksDBKeyedStateBackendBuilder.java:284) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:587) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.contrib.streaming.state.RocksDBStateBackend.createKeyedStateBackend(RocksDBStateBackend.java:93) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.lambda$keyedStatedBackend$1(StreamTaskStateInitializerImpl.java:328) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.attemptCreateAndRestore(BackendRestorerProcedure.java:168) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.BackendRestorerProcedure.createAndRestore(BackendRestorerProcedure.java:135) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.keyedStatedBackend(StreamTaskStateInitializerImpl.java:345) ~[zdata-flink-streams.jar:0.1] at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTaskStateInitializerImpl.java:163) ~[zdata-flink-streams.jar:0.1] ... 9 more

 

A RocksDb state backend is being used along with valueState.