@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 testMapStateReturnsEmptyMapByDefault() throws Exception {

        StreamingRuntimeContext context = createMapOperatorRuntimeContext();

        MapStateDescriptor<Integer, String> descr =
                new MapStateDescriptor<>("name", Integer.class, String.class);
        MapState<Integer, String> state = context.getMapState(descr);

        Iterable<Map.Entry<Integer, String>> value = state.entries();
        assertThat(value).isNotNull();
        assertThat(value.iterator()).isExhausted();
    }