[FLINK-25494] Duplicate element serializer during DefaultOperatorStateBackendSnapshotStrategy#syncPrepareResources

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Invalid
• Fix Version/s: 1.13.6, 1.14.4, 1.15.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Yun Tang
• Reporter: Yun Tang
• Created: 31/Dec/21 06:42
• Updated: 19/Jan/22 03:12
• Resolved: 19/Jan/22 03:12

Description

Currently, during DefaultOperatorStateBackendSnapshotStrategy#syncPrepareResources, it will copy the array list serializer via PartitionableListState#deepCopy. However, it just initialize another ArrayListSerializer and not duplicate the internal state serializer:

 

See "internalListCopySerializer":
 

private PartitionableListState(
        RegisteredOperatorStateBackendMetaInfo<S> stateMetaInfo, ArrayList<S> internalList) {

    this.stateMetaInfo = Preconditions.checkNotNull(stateMetaInfo);
    this.internalList = Preconditions.checkNotNull(internalList);
    this.internalListCopySerializer =
            new ArrayListSerializer<>(stateMetaInfo.getPartitionStateSerializer());
} 

 

This would cause unexpected problem with the usage of kryo serializer.

Issue Links

• links to: GitHub Pull Request #18387