[
https://issues.apache.org/jira/browse/FLINK-27218?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yue Ma updated FLINK-27218:
---------------------------
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, 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.
Therefore, when we update the StateMetaInfo, the internalListCopySerializer
also needs to be updated.
This problem also exists in BroadcastState
was:
Now OperatorState such as can only be constructed *BroadcastState* or
*PartitionableListState* 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.
Therefore, when we update the StateMetaInfo, the internalListCopySerializer
also needs to be updated.
This problem also exists in BroadcastState
> 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
> Priority: Major
> Attachments: image-2022-04-13-14-50-10-921.png
>
>
> 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, 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.
> 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)
