@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();
	// check that the Path class is really registered, i.e., the execution config was applied
	assertTrue(valueSerializer instanceof KryoSerializer);
	assertTrue(((KryoSerializer<?>) valueSerializer).getKryo().getRegistration(Path.class).getId() > 0);
}