@Override
    public <UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties) {
        KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties);
        stateProperties.initializeSerializerUnlessSet(this::createSerializer);
        return keyedStateStore.getMapState(stateProperties);
    }
@Test
    void testMapStateInstantiation() throws Exception {

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

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

        StreamingRuntimeContext context = createRuntimeContext(descriptorCapture, config);

        MapStateDescriptor<String, TaskInfo> descr =
                new MapStateDescriptor<>("name", String.class, TaskInfo.class);

        context.getMapState(descr);

        MapStateDescriptor<?, ?> descrIntercepted =
                (MapStateDescriptor<?, ?>) descriptorCapture.get();
        TypeSerializer<?> valueSerializer = descrIntercepted.getValueSerializer();

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