StreamingJobGraphGenerator { public JobGraph createJobGraph() { jobGraph = new JobGraph(streamGraph.getJobName()); jobGraph.setScheduleMode(ScheduleMode.ALL); init(); Map<Integer, byte[]> hashes = traverseStreamGraphAndGenerateHashes(); setChaining(hashes); setPhysicalEdges(); setSlotSharing(); configureCheckpointing(); configureRestartStrategy(); try { InstantiationUtil.writeObjectToConfig(this.streamGraph.getExecutionConfig(), this.jobGraph.getJobConfiguration(), ExecutionConfig.CONFIG_KEY); } catch (IOException e) { throw new RuntimeException("Config object could not be written to Job Configuration: ", e); } return jobGraph; } StreamingJobGraphGenerator(StreamGraph streamGraph); JobGraph createJobGraph();  }
@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()); }