@Test
public void testExecutionConfigSerialization() throws IOException, ClassNotFoundException {

	final long seed = System.currentTimeMillis();
	final Random r = new Random(seed);
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	StreamGraph streamingJob = new StreamGraph(env);
	StreamingJobGraphGenerator compiler = new StreamingJobGraphGenerator(streamingJob);
	
	boolean closureCleanerEnabled = r.nextBoolean(), forceAvroEnabled = r.nextBoolean(), forceKryoEnabled = r.nextBoolean(), objectReuseEnabled = r.nextBoolean(), sysoutLoggingEnabled = r.nextBoolean();
	int dop = 1 + r.nextInt(10);
	
	ExecutionConfig config = streamingJob.getExecutionConfig();
	if(closureCleanerEnabled) {
		config.enableClosureCleaner();
	} else {
		config.disableClosureCleaner();
	}
	if(forceAvroEnabled) {
		config.enableForceAvro();
	} else {
		config.disableForceAvro();
	}
	if(forceKryoEnabled) {
		config.enableForceKryo();
	} else {
		config.disableForceKryo();
	}
	if(objectReuseEnabled) {
		config.enableObjectReuse();
	} else {
		config.disableObjectReuse();
	}
	if(sysoutLoggingEnabled) {
		config.enableSysoutLogging();
	} else {
		config.disableSysoutLogging();
	}
	config.setParallelism(dop);
	
	JobGraph jobGraph = compiler.createJobGraph();
	
	ExecutionConfig executionConfig = InstantiationUtil.readObjectFromConfig(
			jobGraph.getJobConfiguration(),
			ExecutionConfig.CONFIG_KEY,
			Thread.currentThread().getContextClassLoader());
	
	assertNotNull(executionConfig);
	
	assertEquals(closureCleanerEnabled, executionConfig.isClosureCleanerEnabled());
	assertEquals(forceAvroEnabled, executionConfig.isForceAvroEnabled());
	assertEquals(forceKryoEnabled, executionConfig.isForceKryoEnabled());
	assertEquals(objectReuseEnabled, executionConfig.isObjectReuseEnabled());
	assertEquals(sysoutLoggingEnabled, executionConfig.isSysoutLoggingEnabled());
	assertEquals(dop, executionConfig.getParallelism());
}