[FLINK-29645] BatchExecutionKeyedStateBackend is using incorrect ExecutionConfig when creating serializer

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.12.7, 1.13.6, 1.16.0, 1.17.0, 1.15.2, 1.14.6
• Fix Version/s: 1.16.0, 1.17.0, 1.15.3
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Dawid Wysakowicz
• Reporter: Piotr Nowojski
• Created: 14/Oct/22 13:22
• Updated: 17/Oct/22 15:27
• Resolved: 17/Oct/22 15:27

Description

org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionKeyedStateBackend#getOrCreateKeyedState is using freshly constructed ExecutionConfig, instead of the one configured by the user from the environment.

    public <N, S extends State, T> S getOrCreateKeyedState(
            TypeSerializer<N> namespaceSerializer, StateDescriptor<S, T> stateDescriptor)
            throws Exception {
        checkNotNull(namespaceSerializer, "Namespace serializer");
        checkNotNull(
                keySerializer,
                "State key serializer has not been configured in the config. "
                        + "This operation cannot use partitioned state.");

        if (!stateDescriptor.isSerializerInitialized()) {
            stateDescriptor.initializeSerializerUnlessSet(new ExecutionConfig());
        }

The correct one could be obtained from env.getExecutionConfig() in org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionStateBackend#createKeyedStateBackend

Issue Links

• links to: GitHub Pull Request #21082