@Override
    public <T> ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties) {
        KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
        stateProperties.initializeSerializerUnlessSet(this::createSerializer);
        return keyedStateStore.getReducingState(stateProperties);
    }
@Test
    void testReducingStateInstantiation() throws Exception {

        final ExecutionConfig config = new ExecutionConfig();
        config.getSerializerConfig().registerKryoType(Path.class);

        final AtomicReference<Object> descriptorCapture = new AtomicReference<>();

        StreamingRuntimeContext context = createRuntimeContext(descriptorCapture, config);

        @SuppressWarnings("unchecked")
        ReduceFunction<TaskInfo> reducer = (ReduceFunction<TaskInfo>) mock(ReduceFunction.class);

        ReducingStateDescriptor<TaskInfo> descr =
                new ReducingStateDescriptor<>("name", reducer, TaskInfo.class);

        context.getReducingState(descr);

        StateDescriptor<?, ?> descrIntercepted = (StateDescriptor<?, ?>) descriptorCapture.get();
        TypeSerializer<?> serializer = descrIntercepted.getSerializer();

        // check that the Path class is really registered, i.e., the execution config was applied
        assertThat(serializer).isInstanceOf(KryoSerializer.class);
        assertThat(((KryoSerializer<?>) serializer).getKryo().getRegistration(Path.class).getId())
                .isPositive();
    }