import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@Test
public void testDeployerWithIsolatedConfiguration() throws Exception {
  YarnClusterConfiguration clusterConf = mock(YarnClusterConfiguration.class);
  doReturn(new YarnConfiguration()).when(clusterConf).conf();
  ScheduledExecutorService executor = mock(ScheduledExecutorService.class);
  Configuration flinkConf = new Configuration();
  YarnClient client = mock(YarnClient.class);
  JobDeployer deploy = new JobDeployer(clusterConf, client, executor, flinkConf);
  AthenaXYarnClusterDescriptor desc = mock(AthenaXYarnClusterDescriptor.class);

  YarnClusterClient clusterClient = mock(YarnClusterClient.class);
  doReturn(clusterClient).when(desc).deploy();

  ActorGateway actorGateway = mock(ActorGateway.class);
  doReturn(actorGateway).when(clusterClient).getJobManagerGateway();
  doReturn(Future$.MODULE$.successful(null)).when(actorGateway).ask(any(), any());

  JobGraph jobGraph = mock(JobGraph.class);
  doReturn(JobID.generate()).when(jobGraph).getJobID();
  deploy.start(desc, jobGraph);

  ArgumentCaptor<Configuration> args = ArgumentCaptor.forClass(Configuration.class);
  verify(desc).setFlinkConfiguration(args.capture());
  assertNotSame(flinkConf, args.getValue());
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
private JobGraph getJobGraph(FlinkPlan optPlan, List<URL> jarFiles, List<URL> classpaths, SavepointRestoreSettings savepointSettings) {
	JobGraph job;
	if (optPlan instanceof StreamingPlan) {
		job = ((StreamingPlan) optPlan).getJobGraph();
		job.setSavepointRestoreSettings(savepointSettings);
	} else {
		JobGraphGenerator gen = new JobGraphGenerator(this.flinkConfig);
		job = gen.compileJobGraph((OptimizedPlan) optPlan);
	}

	for (URL jar : jarFiles) {
		try {
			job.addJar(new Path(jar.toURI()));
		} catch (URISyntaxException e) {
			throw new RuntimeException("URL is invalid. This should not happen.", e);
		}
	}

	job.setClasspaths(classpaths);

	return job;
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@BeforeClass
public static void setup() {
	try {
		Configuration config = new Configuration();
		config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 4L);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_SLOTS / 2);

		cluster = new LocalFlinkMiniCluster(config);

		cluster.start();

		final JobVertex jobVertex = new JobVertex("Working job vertex.");
		jobVertex.setInvokableClass(NoOpInvokable.class);
		workingJobGraph = new JobGraph("Working testing job", jobVertex);
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * Tests that in a streaming use case where checkpointing is enabled, a
 * fixed delay with Integer.MAX_VALUE retries is instantiated if no other restart
 * strategy has been specified.
 */
@Test
public void testAutomaticRestartingWhenCheckpointing() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.enableCheckpointing(500);

	env.fromElements(1).print();

	StreamGraph graph = env.getStreamGraph();
	JobGraph jobGraph = graph.getJobGraph();

	RestartStrategies.RestartStrategyConfiguration restartStrategy =
		jobGraph.getSerializedExecutionConfig().deserializeValue(getClass().getClassLoader()).getRestartStrategy();

	Assert.assertNotNull(restartStrategy);
	Assert.assertTrue(restartStrategy instanceof RestartStrategies.FixedDelayRestartStrategyConfiguration);
	Assert.assertEquals(Integer.MAX_VALUE, ((RestartStrategies.FixedDelayRestartStrategyConfiguration) restartStrategy).getRestartAttempts());
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * Checks that in a streaming use case where checkpointing is enabled and the number
 * of execution retries is set to 42 and the delay to 1337, fixed delay restarting is used.
 */
@Test
public void testFixedRestartingWhenCheckpointingAndExplicitExecutionRetriesNonZero() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.enableCheckpointing(500);
	env.setNumberOfExecutionRetries(42);
	env.getConfig().setExecutionRetryDelay(1337);

	env.fromElements(1).print();

	StreamGraph graph = env.getStreamGraph();
	JobGraph jobGraph = graph.getJobGraph();

	RestartStrategies.RestartStrategyConfiguration restartStrategy =
		jobGraph.getSerializedExecutionConfig().deserializeValue(getClass().getClassLoader()).getRestartStrategy();

	Assert.assertNotNull(restartStrategy);
	Assert.assertTrue(restartStrategy instanceof RestartStrategies.FixedDelayRestartStrategyConfiguration);
	Assert.assertEquals(42, ((RestartStrategies.FixedDelayRestartStrategyConfiguration) restartStrategy).getRestartAttempts());
	Assert.assertEquals(1337, ((RestartStrategies.FixedDelayRestartStrategyConfiguration) restartStrategy).getDelayBetweenAttemptsInterval().toMilliseconds());
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
private JobGraph createJobGraph(ExecutionMode mode) {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
	env.enableCheckpointing(500, CheckpointingMode.EXACTLY_ONCE);
	env.setRestartStrategy(RestartStrategies.noRestart());
	env.setStateBackend(new MemoryStateBackend());

	switch (mode) {
		case MIGRATE:
			createMigrationJob(env);
			break;
		case RESTORE:
			createRestoredJob(env);
			break;
	}

	return StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@BeforeClass
public static void setup() {
	try {
		Configuration config = new Configuration();
		config.setLong(TaskManagerOptions.MANAGED_MEMORY_SIZE, 4L);
		config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
		config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_SLOTS / 2);
		
		cluster = new LocalFlinkMiniCluster(config);

		cluster.start();
		
		final JobVertex jobVertex = new JobVertex("Working job vertex.");
		jobVertex.setInvokableClass(NoOpInvokable.class);
		workingJobGraph = new JobGraph("Working testing job", jobVertex);
	}
	catch (Exception e) {
		e.printStackTrace();
		fail(e.getMessage());
	}
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * Tests that all operations work and don't change the state.
 */
@Test
public void testNoOps() throws Exception {
	StandaloneSubmittedJobGraphStore jobGraphs = new StandaloneSubmittedJobGraphStore();

	SubmittedJobGraph jobGraph = new SubmittedJobGraph(
			new JobGraph("testNoOps"),
			new JobInfo(ActorRef.noSender(), ListeningBehaviour.DETACHED, 0, Integer.MAX_VALUE));

	assertEquals(0, jobGraphs.getJobIds().size());

	jobGraphs.putJobGraph(jobGraph);
	assertEquals(0, jobGraphs.getJobIds().size());

	jobGraphs.removeJobGraph(jobGraph.getJobGraph().getJobID());
	assertEquals(0, jobGraphs.getJobIds().size());

	assertNull(jobGraphs.recoverJobGraph(new JobID()));
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
	ParameterTool params = ParameterTool.fromArgs(args);

	// define the dataflow
	StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
	env.setParallelism(2);
	env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, 1000));
	env.readFileStream("input/", 60000, FileMonitoringFunction.WatchType.ONLY_NEW_FILES)
		.addSink(new DiscardingSink<String>());

	// generate a job graph
	final JobGraph jobGraph = env.getStreamGraph().getJobGraph();
	File jobGraphFile = new File(params.get("output", "job.graph"));
	try (FileOutputStream output = new FileOutputStream(jobGraphFile);
		ObjectOutputStream obOutput = new ObjectOutputStream(output)){
		obOutput.writeObject(jobGraph);
	}
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@Override
public JobExecutionResult execute(String jobName) throws Exception {
	Plan plan = createProgramPlan(jobName);

	Optimizer pc = new Optimizer(new Configuration());
	OptimizedPlan op = pc.compile(plan);

	JobGraphGenerator jgg = new JobGraphGenerator();
	JobGraph jobGraph = jgg.compileJobGraph(op);

	String jsonPlan = JsonPlanGenerator.generatePlan(jobGraph);

	// first check that the JSON is valid
	JsonParser parser = new JsonFactory().createJsonParser(jsonPlan);
	while (parser.nextToken() != null) {}

	validator.validateJson(jsonPlan);

	throw new AbortError();
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * Tests that there are no collisions with two identical intermediate nodes connected to the
 * same predecessor.
 *
 * <pre>
 *             /-> [ (map) ] -> [ (sink) ]
 * [ (src) ] -+
 *             \-> [ (map) ] -> [ (sink) ]
 * </pre>
 */
@Test
public void testNodeHashIdenticalNodes() throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
	env.setParallelism(4);
	env.disableOperatorChaining();

	DataStream<String> src = env.addSource(new NoOpSourceFunction());

	src.map(new NoOpMapFunction()).addSink(new NoOpSinkFunction());

	src.map(new NoOpMapFunction()).addSink(new NoOpSinkFunction());

	JobGraph jobGraph = env.getStreamGraph().getJobGraph();
	Set<JobVertexID> vertexIds = new HashSet<>();
	for (JobVertex vertex : jobGraph.getVertices()) {
		assertTrue(vertexIds.add(vertex.getID()));
	}
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@Override
public CompletableFuture<String> handleJsonRequest(Map<String, String> pathParams, Map<String, String> queryParams, JobManagerGateway jobManagerGateway) {
	return CompletableFuture.supplyAsync(
		() -> {
			try {
				JarActionHandlerConfig config = JarActionHandlerConfig.fromParams(pathParams, queryParams);
				JobGraph graph = getJobGraphAndClassLoader(config).f0;
				StringWriter writer = new StringWriter();
				JsonGenerator gen = JsonFactory.JACKSON_FACTORY.createGenerator(writer);
				gen.writeStartObject();
				gen.writeFieldName("plan");
				gen.writeRawValue(JsonPlanGenerator.generatePlan(graph));
				gen.writeEndObject();
				gen.close();
				return writer.toString();
			}
			catch (Exception e) {
				throw new CompletionException(e);
			}
		},
		executor);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
private JobManagerRunner createJobManagerRunner(Configuration config) throws Exception{
	// first get JobGraph from local resources
	//TODO: generate the job graph from user's jar
	JobGraph jobGraph = loadJobGraph(config);

	// now the JobManagerRunner
	return new JobManagerRunner(
		ResourceID.generate(),
		jobGraph,
		config,
		commonRpcService,
		haServices,
		heartbeatServices,
		this,
		this);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * This method runs a job in blocking mode. The method returns only after the job
 * completed successfully, or after it failed terminally.
 *
 * @param job  The Flink job to execute 
 * @return The result of the job execution
 *
 * @throws JobExecutionException Thrown if anything went amiss during initial job launch,
 *         or if the job terminally failed.
 */
@Override
public JobExecutionResult executeJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
	checkNotNull(job, "job is null");

	MiniClusterJobDispatcher dispatcher;
	synchronized (lock) {
		checkState(running, "mini cluster is not running");
		dispatcher = this.jobDispatcher;
	}

	// we have to allow queued scheduling in Flip-6 mode because we need to request slots
	// from the ResourceManager
	job.setAllowQueuedScheduling(true);

	return dispatcher.runJobBlocking(job);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
/**
 * This method runs a job in blocking mode. The method returns only after the job
 * completed successfully, or after it failed terminally.
 *
 * @param job  The Flink job to execute 
 * @return The result of the job execution
 *
 * @throws JobExecutionException Thrown if anything went amiss during initial job launch,
 *         or if the job terminally failed.
 */
public JobExecutionResult runJobBlocking(JobGraph job) throws JobExecutionException, InterruptedException {
	checkNotNull(job);
	
	LOG.info("Received job for blocking execution: {} ({})", job.getName(), job.getJobID());
	final BlockingJobSync sync = new BlockingJobSync(job.getJobID(), numJobManagers);

	synchronized (lock) {
		checkState(!shutdown, "mini cluster is shut down");
		checkState(runners == null, "mini cluster can only execute one job at a time");

		this.runners = startJobRunners(job, sync, sync);
	}

	try {
		return sync.getResult();
	}
	finally {
		// always clear the status for the next job
		runners = null;
		clearJobRunningState(job.getJobID());
	}
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
private StreamingJobGraphGenerator(StreamGraph streamGraph) {
	this.streamGraph = streamGraph;
	this.defaultStreamGraphHasher = new StreamGraphHasherV2();
	this.legacyStreamGraphHashers = Arrays.asList(new StreamGraphUserHashHasher());

	this.jobVertices = new HashMap<>();
	this.builtVertices = new HashSet<>();
	this.chainedConfigs = new HashMap<>();
	this.vertexConfigs = new HashMap<>();
	this.chainedNames = new HashMap<>();
	this.chainedMinResources = new HashMap<>();
	this.chainedPreferredResources = new HashMap<>();
	this.physicalEdgesInOrder = new ArrayList<>();

	jobGraph = new JobGraph(streamGraph.getJobName());
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
@Override
protected JobManagerRunner createJobManagerRunner(
		ResourceID resourceId,
		JobGraph jobGraph,
		Configuration configuration,
		RpcService rpcService,
		HighAvailabilityServices highAvailabilityServices,
		HeartbeatServices heartbeatServices,
		JobManagerServices jobManagerServices,
		MetricRegistry metricRegistry,
		OnCompletionActions onCompleteActions,
		FatalErrorHandler fatalErrorHandler) throws Exception {
	// create the standard job manager runner
	return new JobManagerRunner(
		resourceId,
		jobGraph,
		configuration,
		rpcService,
		highAvailabilityServices,
		heartbeatServices,
		jobManagerServices,
		metricRegistry,
		onCompleteActions,
		fatalErrorHandler,
		restAddress);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
protected JobManagerRunner createJobManagerRunner(
		Configuration configuration,
		ResourceID resourceId,
		RpcService rpcService,
		HighAvailabilityServices highAvailabilityServices,
		JobManagerServices jobManagerServices,
		HeartbeatServices heartbeatServices,
		MetricRegistry metricRegistry,
		FatalErrorHandler fatalErrorHandler,
		@Nullable String restAddress) throws Exception {

	final JobGraph jobGraph = retrieveJobGraph(configuration);

	return new JobManagerRunner(
		resourceId,
		jobGraph,
		configuration,
		rpcService,
		highAvailabilityServices,
		heartbeatServices,
		jobManagerServices,
		metricRegistry,
		new TerminatingOnCompleteActions(jobGraph.getJobID()),
		fatalErrorHandler,
		restAddress);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
public JobGraph createBlockingJob(int parallelism) {
	Tasks.BlockingOnceReceiver$.MODULE$.blocking_$eq(true);

	JobVertex sender = new JobVertex("sender");
	JobVertex receiver = new JobVertex("receiver");

	sender.setInvokableClass(Tasks.Sender.class);
	receiver.setInvokableClass(Tasks.BlockingOnceReceiver.class);

	sender.setParallelism(parallelism);
	receiver.setParallelism(parallelism);

	receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED);

	SlotSharingGroup slotSharingGroup = new SlotSharingGroup();
	sender.setSlotSharingGroup(slotSharingGroup);
	receiver.setSlotSharingGroup(slotSharingGroup);

	return new JobGraph("Blocking test job", sender, receiver);
}
 

import org.apache.flink.runtime.jobgraph.JobGraph; //导入依赖的package包/类
private static JobGraph getSimpleJob() throws IOException {
	JobVertex task = new JobVertex("Test task");
	task.setParallelism(1);
	task.setMaxParallelism(1);
	task.setInvokableClass(NoOpInvokable.class);

	JobGraph jg = new JobGraph(new JobID(), "Test Job", task);
	jg.setAllowQueuedScheduling(true);
	jg.setScheduleMode(ScheduleMode.EAGER);

	ExecutionConfig executionConfig = new ExecutionConfig();
	executionConfig.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000));
	jg.setExecutionConfig(executionConfig);

	return jg;
}