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

[Yue Ma (Jira)]

Yue Ma created FLINK-27218:
------------------------------

             Summary: Serializer in OperatorState has not been updated when new 
Serializers are NOT incompatible
                 Key: FLINK-27218
                 URL: https://issues.apache.org/jira/browse/FLINK-27218
             Project: Flink
          Issue Type: Bug
          Components: Runtime / State Backends
    Affects Versions: 1.15.1
            Reporter: Yue Ma
         Attachments: image-2022-04-13-14-50-10-921.png

Now OperatorState can only be constructed via DefaultOperatorStateBackend. But 
when *BroadcastState* or *PartitionableListState* Serializer changes, it seems 
to have the following problems.

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

First, we will construct a restored PartitionableListState based on the 
information in the snapshot during the restoreOperation.

Then we will update the StateMetaInfo in partitionableListState as the 
following code

 
{code:java}
TypeSerializerSchemaCompatibility<S> stateCompatibility =
                
restoredPartitionableListStateMetaInfo.updatePartitionStateSerializer(newPartitionStateSerializer);

partitionableListState.setStateMetaInfo(restoredPartitionableListStateMetaInfo);{code}
 


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. 
But internalListCopySerializer will be used later when making the snopshot. 
Therefore, when we update the StateMetaInfo, the internalListCopySerializer 
also needs to be updated.

This problem also exists in BroadcastState



--
This message was sent by Atlassian Jira
(v8.20.1#820001)