[FLINK-27218] Serializer in OperatorState has not been updated when new Serializers are NOT incompatible

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.15.1
• Fix Version/s: 1.15.1, 1.16.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Yue Ma
• Reporter: Yue Ma
• Created: 13/Apr/22 07:11
• Updated: 22/Apr/22 14:44
• Resolved: 22/Apr/22 14:44

Description

OperatorState such as BroadcastState or PartitionableListState  can only be constructed via DefaultOperatorStateBackend. But when BroadcastState or PartitionableListState Serializer changes after we restart the job , it seems to have the following problems .

As an example, we can see how PartitionableListState is initialized.

First, RestoreOperation will construct a restored PartitionableListState based on the information in the snapshot.

Then StateMetaInfo in partitionableListState will be updated  as the following code

TypeSerializerSchemaCompatibility<S> stateCompatibility =
                restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);

partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);

The main problem is that there is also an internalListCopySerializer in PartitionableListState that is built using the previous Serializer and it has not been updated. 

Therefore, when we update the StateMetaInfo, the internalListCopySerializer also needs to be updated.

 

Issue Links

• links to: GitHub Pull Request #19508