### Question:
StreamingRuntimeContext extends AbstractRuntimeUDFContext { @Override public <UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties) { KeyedStateStore keyedStateStore = checkPreconditionsAndGetKeyedStateStore(stateProperties); stateProperties.initializeSerializerUnlessSet(getExecutionConfig()); return keyedStateStore.getMapState(stateProperties); } StreamingRuntimeContext(AbstractStreamOperator<?> operator,
									Environment env, Map<String, Accumulator<?, ?>> accumulators); InputSplitProvider getInputSplitProvider(); ProcessingTimeService getProcessingTimeService(); GlobalAggregateManager getGlobalAggregateManager(); String getOperatorUniqueID(); @Override boolean hasBroadcastVariable(String name); @Override List<RT> getBroadcastVariable(String name); @Override C getBroadcastVariableWithInitializer(String name, BroadcastVariableInitializer<T, C> initializer); @Override ValueState<T> getState(ValueStateDescriptor<T> stateProperties); @Override ListState<T> getListState(ListStateDescriptor<T> stateProperties); @Override ReducingState<T> getReducingState(ReducingStateDescriptor<T> stateProperties); @Override AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT> stateProperties); @Override FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC> stateProperties); @Override MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV> stateProperties); boolean isCheckpointingEnabled(); CheckpointingMode getCheckpointMode(); long getBufferTimeout();  }### Answer:
@Test public void testMapStateInstantiation() throws Exception { final ExecutionConfig config = new ExecutionConfig(); config.registerKryoType(Path.class); final AtomicReference<Object> descriptorCapture = new AtomicReference<>(); StreamingRuntimeContext context = new StreamingRuntimeContext( createDescriptorCapturingMockOp(descriptorCapture, config), createMockEnvironment(), Collections.<String, Accumulator<?, ?>>emptyMap()); MapStateDescriptor<String, TaskInfo> descr = new MapStateDescriptor<>("name", String.class, TaskInfo.class); context.getMapState(descr); MapStateDescriptor<?, ?> descrIntercepted = (MapStateDescriptor<?, ?>) descriptorCapture.get(); TypeSerializer<?> valueSerializer = descrIntercepted.getValueSerializer(); assertTrue(valueSerializer instanceof KryoSerializer); assertTrue(((KryoSerializer<?>) valueSerializer).getKryo().getRegistration(Path.class).getId() > 0); }
@Test public void testMapStateReturnsEmptyMapByDefault() throws Exception { StreamingRuntimeContext context = new StreamingRuntimeContext( createMapPlainMockOp(), createMockEnvironment(), Collections.<String, Accumulator<?, ?>>emptyMap()); 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(); assertNotNull(value); assertFalse(value.iterator().hasNext()); }