test_class.file: flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
test_class.identifier: StreamingJobGraphGeneratorTest
test_case.body: @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());
  	}
test_case.class_method_signature: StreamingJobGraphGeneratorTest.testExecutionConfigSerialization()
test_case.full_signature: @Test public void testExecutionConfigSerialization()
test_case.identifier: testExecutionConfigSerialization
test_case.invocations: currentTimeMillis getExecutionEnvironment nextBoolean nextBoolean nextBoolean nextBoolean nextBoolean nextInt getExecutionConfig enableClosureCleaner disableClosureCleaner enableForceAvro disableForceAvro enableForceKryo disableForceKryo enableObjectReuse disableObjectReuse enableSysoutLogging disableSysoutLogging setParallelism createJobGraph readObjectFromConfig getJobConfiguration getContextClassLoader currentThread assertNotNull assertEquals isClosureCleanerEnabled assertEquals isForceAvroEnabled assertEquals isForceKryoEnabled assertEquals isObjectReuseEnabled assertEquals isSysoutLoggingEnabled assertEquals getParallelism
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testExecutionConfigSerialization()
focal_class.declarator: LOG = LoggerFactory.getLogger(StreamingJobGraphGenerator.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(StreamingJobGraphGenerator.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: DEFAULT_RESTART_DELAY = 10000L
focal_class.modifier: private static final
focal_class.original_string: private static final long DEFAULT_RESTART_DELAY = 10000L;
focal_class.type: long
focal_class.var_name: DEFAULT_RESTART_DELAY
focal_class.declarator: streamGraph
focal_class.modifier: private
focal_class.original_string: private StreamGraph streamGraph;
focal_class.type: StreamGraph
focal_class.var_name: streamGraph
focal_class.declarator: jobVertices
focal_class.modifier: private
focal_class.original_string: private Map<Integer, JobVertex> jobVertices;
focal_class.type: Map<Integer, JobVertex>
focal_class.var_name: jobVertices
focal_class.declarator: jobGraph
focal_class.modifier: private
focal_class.original_string: private JobGraph jobGraph;
focal_class.type: JobGraph
focal_class.var_name: jobGraph
focal_class.declarator: builtVertices
focal_class.modifier: private
focal_class.original_string: private Collection<Integer> builtVertices;
focal_class.type: Collection<Integer>
focal_class.var_name: builtVertices
focal_class.declarator: physicalEdgesInOrder
focal_class.modifier: private
focal_class.original_string: private List<StreamEdge> physicalEdgesInOrder;
focal_class.type: List<StreamEdge>
focal_class.var_name: physicalEdgesInOrder
focal_class.declarator: chainedConfigs
focal_class.modifier: private
focal_class.original_string: private Map<Integer, Map<Integer, StreamConfig>> chainedConfigs;
focal_class.type: Map<Integer, Map<Integer, StreamConfig>>
focal_class.var_name: chainedConfigs
focal_class.declarator: vertexConfigs
focal_class.modifier: private
focal_class.original_string: private Map<Integer, StreamConfig> vertexConfigs;
focal_class.type: Map<Integer, StreamConfig>
focal_class.var_name: vertexConfigs
focal_class.declarator: chainedNames
focal_class.modifier: private
focal_class.original_string: private Map<Integer, String> chainedNames;
focal_class.type: Map<Integer, String>
focal_class.var_name: chainedNames
focal_class.file: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
focal_class.identifier: StreamingJobGraphGenerator
focal_class.class_method_signature: StreamingJobGraphGenerator.StreamingJobGraphGenerator(StreamGraph streamGraph)
focal_class.full_signature: public  StreamingJobGraphGenerator(StreamGraph streamGraph)
focal_class.identifier: StreamingJobGraphGenerator
focal_class.modifiers: public
focal_class.parameters: (StreamGraph streamGraph)
focal_class.signature:  StreamingJobGraphGenerator(StreamGraph streamGraph)
focal_class.class_method_signature: StreamingJobGraphGenerator.init()
focal_class.full_signature: private void init()
focal_class.identifier: init
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void init()
focal_class.class_method_signature: StreamingJobGraphGenerator.createJobGraph()
focal_class.full_signature: public JobGraph createJobGraph()
focal_class.identifier: createJobGraph
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: JobGraph
focal_class.signature: JobGraph createJobGraph()
focal_class.class_method_signature: StreamingJobGraphGenerator.setPhysicalEdges()
focal_class.full_signature: private void setPhysicalEdges()
focal_class.identifier: setPhysicalEdges
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void setPhysicalEdges()
focal_class.class_method_signature: StreamingJobGraphGenerator.setChaining(Map<Integer, byte[]> hashes)
focal_class.full_signature: private void setChaining(Map<Integer, byte[]> hashes)
focal_class.identifier: setChaining
focal_class.modifiers: private
focal_class.parameters: (Map<Integer, byte[]> hashes)
focal_class.return: void
focal_class.signature: void setChaining(Map<Integer, byte[]> hashes)
focal_class.class_method_signature: StreamingJobGraphGenerator.createChain(
  			Integer startNodeId,
  			Integer currentNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.full_signature: private List<StreamEdge> createChain(
  			Integer startNodeId,
  			Integer currentNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.identifier: createChain
focal_class.modifiers: private
focal_class.parameters: (
  			Integer startNodeId,
  			Integer currentNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.return: List<StreamEdge>
focal_class.signature: List<StreamEdge> createChain(
  			Integer startNodeId,
  			Integer currentNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.class_method_signature: StreamingJobGraphGenerator.createChainedName(Integer vertexID, List<StreamEdge> chainedOutputs)
focal_class.full_signature: private String createChainedName(Integer vertexID, List<StreamEdge> chainedOutputs)
focal_class.identifier: createChainedName
focal_class.modifiers: private
focal_class.parameters: (Integer vertexID, List<StreamEdge> chainedOutputs)
focal_class.return: String
focal_class.signature: String createChainedName(Integer vertexID, List<StreamEdge> chainedOutputs)
focal_class.class_method_signature: StreamingJobGraphGenerator.createJobVertex(
  			Integer streamNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.full_signature: private StreamConfig createJobVertex(
  			Integer streamNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.identifier: createJobVertex
focal_class.modifiers: private
focal_class.parameters: (
  			Integer streamNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.return: StreamConfig
focal_class.signature: StreamConfig createJobVertex(
  			Integer streamNodeId,
  			Map<Integer, byte[]> hashes)
focal_class.class_method_signature: StreamingJobGraphGenerator.setVertexConfig(Integer vertexID, StreamConfig config,
  			List<StreamEdge> chainableOutputs, List<StreamEdge> nonChainableOutputs)
focal_class.full_signature: @SuppressWarnings("unchecked") private void setVertexConfig(Integer vertexID, StreamConfig config,
  			List<StreamEdge> chainableOutputs, List<StreamEdge> nonChainableOutputs)
focal_class.identifier: setVertexConfig
focal_class.modifiers: @SuppressWarnings("unchecked") private
focal_class.parameters: (Integer vertexID, StreamConfig config,
  			List<StreamEdge> chainableOutputs, List<StreamEdge> nonChainableOutputs)
focal_class.return: void
focal_class.signature: void setVertexConfig(Integer vertexID, StreamConfig config,
  			List<StreamEdge> chainableOutputs, List<StreamEdge> nonChainableOutputs)
focal_class.class_method_signature: StreamingJobGraphGenerator.connect(Integer headOfChain, StreamEdge edge)
focal_class.full_signature: private void connect(Integer headOfChain, StreamEdge edge)
focal_class.identifier: connect
focal_class.modifiers: private
focal_class.parameters: (Integer headOfChain, StreamEdge edge)
focal_class.return: void
focal_class.signature: void connect(Integer headOfChain, StreamEdge edge)
focal_class.class_method_signature: StreamingJobGraphGenerator.isChainable(StreamEdge edge)
focal_class.full_signature: private boolean isChainable(StreamEdge edge)
focal_class.identifier: isChainable
focal_class.modifiers: private
focal_class.parameters: (StreamEdge edge)
focal_class.return: boolean
focal_class.signature: boolean isChainable(StreamEdge edge)
focal_class.class_method_signature: StreamingJobGraphGenerator.setSlotSharing()
focal_class.full_signature: private void setSlotSharing()
focal_class.identifier: setSlotSharing
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void setSlotSharing()
focal_class.class_method_signature: StreamingJobGraphGenerator.configureCheckpointing()
focal_class.full_signature: private void configureCheckpointing()
focal_class.identifier: configureCheckpointing
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void configureCheckpointing()
focal_class.class_method_signature: StreamingJobGraphGenerator.configureRestartStrategy()
focal_class.full_signature: private void configureRestartStrategy()
focal_class.identifier: configureRestartStrategy
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void configureRestartStrategy()
focal_class.class_method_signature: StreamingJobGraphGenerator.traverseStreamGraphAndGenerateHashes()
focal_class.full_signature: private Map<Integer, byte[]> traverseStreamGraphAndGenerateHashes()
focal_class.identifier: traverseStreamGraphAndGenerateHashes
focal_class.modifiers: private
focal_class.parameters: ()
focal_class.return: Map<Integer, byte[]>
focal_class.signature: Map<Integer, byte[]> traverseStreamGraphAndGenerateHashes()
focal_class.class_method_signature: StreamingJobGraphGenerator.generateNodeHash(
  			StreamNode node,
  			HashFunction hashFunction,
  			Map<Integer, byte[]> hashes)
focal_class.full_signature: private boolean generateNodeHash(
  			StreamNode node,
  			HashFunction hashFunction,
  			Map<Integer, byte[]> hashes)
focal_class.identifier: generateNodeHash
focal_class.modifiers: private
focal_class.parameters: (
  			StreamNode node,
  			HashFunction hashFunction,
  			Map<Integer, byte[]> hashes)
focal_class.return: boolean
focal_class.signature: boolean generateNodeHash(
  			StreamNode node,
  			HashFunction hashFunction,
  			Map<Integer, byte[]> hashes)
focal_class.class_method_signature: StreamingJobGraphGenerator.generateUserSpecifiedHash(StreamNode node, Hasher hasher)
focal_class.full_signature: private byte[] generateUserSpecifiedHash(StreamNode node, Hasher hasher)
focal_class.identifier: generateUserSpecifiedHash
focal_class.modifiers: private
focal_class.parameters: (StreamNode node, Hasher hasher)
focal_class.return: byte[]
focal_class.signature: byte[] generateUserSpecifiedHash(StreamNode node, Hasher hasher)
focal_class.class_method_signature: StreamingJobGraphGenerator.generateDeterministicHash(
  			StreamNode node,
  			Hasher hasher,
  			Map<Integer, byte[]> hashes)
focal_class.full_signature: private byte[] generateDeterministicHash(
  			StreamNode node,
  			Hasher hasher,
  			Map<Integer, byte[]> hashes)
focal_class.identifier: generateDeterministicHash
focal_class.modifiers: private
focal_class.parameters: (
  			StreamNode node,
  			Hasher hasher,
  			Map<Integer, byte[]> hashes)
focal_class.return: byte[]
focal_class.signature: byte[] generateDeterministicHash(
  			StreamNode node,
  			Hasher hasher,
  			Map<Integer, byte[]> hashes)
focal_class.class_method_signature: StreamingJobGraphGenerator.generateNodeLocalHash(StreamNode node, Hasher hasher, int id)
focal_class.full_signature: private void generateNodeLocalHash(StreamNode node, Hasher hasher, int id)
focal_class.identifier: generateNodeLocalHash
focal_class.modifiers: private
focal_class.parameters: (StreamNode node, Hasher hasher, int id)
focal_class.return: void
focal_class.signature: void generateNodeLocalHash(StreamNode node, Hasher hasher, int id)
focal_method.body: public JobGraph createJobGraph() {
  		jobGraph = new JobGraph(streamGraph.getJobName());
  
  		// make sure that all vertices start immediately
  		jobGraph.setScheduleMode(ScheduleMode.ALL);
  
  		init();
  
  		// Generate deterministic hashes for the nodes in order to identify them across
  		// submission iff they didn't change.
  		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;
  	}
focal_method.class_method_signature: StreamingJobGraphGenerator.createJobGraph()
focal_method.full_signature: public JobGraph createJobGraph()
focal_method.identifier: createJobGraph
focal_method.invocations: getJobName setScheduleMode init traverseStreamGraphAndGenerateHashes setChaining setPhysicalEdges setSlotSharing configureCheckpointing configureRestartStrategy writeObjectToConfig getExecutionConfig getJobConfiguration
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: JobGraph
focal_method.signature: JobGraph createJobGraph()