以flink1.10开源代码未例子。梳理flink client提交作业的源码流程

1、/bin/flink run xxx

用户通过flink client的命令行工具提交作业
flink只是个shell脚本，原理就是调用如下类 org.apache.flink.client.cli.CliFrontend “$@”

2、CliFrontend的main()函数内部逻辑

getConfigurationDirectoryFromEnv(); - 获取配置文件的目录-FLINK_CONF_DIR环境变量
GlobalConfiguration.loadConfiguration(configurationDirectory) - 解析flink-conf.yaml文件
loadCustomCommandLines - 这个是为不同cli解析模式而设计的
cli = new CliFrontend(configuration,customCommandLines);
cli.parseParameters(args)
	run(params) - 根据flink run这个关键参数走run流程

3、run(params)

run(params) - 此流程是client解析出作业流程并提交的主要流程

CliFrontendParser.getRunCommandOptions
CliFrontendParser.parse  -解析参数
program = buildProgram(programOptions);
	dependencyFiles 
	PackagedProgram.newBuilder()
effectiveConfiguration = getEffectiveConfiguration(commandLine, programOptions, jobJars);
executeProgram(effectiveConfiguration, program) - 核心的执行用户自定义的jar作业
	ContextEnvironment.setAsContext(factory); - 设置classloader和上下文环境
	program.invokeInteractiveModeForExecution(); - 用反射执行用户的用户自定义的jar作业的main函数
		callMainMethod
			mainMethod = entryClass.getMethod("main", String[].class);
			mainMethod.invoke(null, (Object) args);
program.deleteExtractedLibraries();

4、核心的ExecutionEnvironment.execute()

用户写 jar作业最后都会调用env.execute()执行这个函数就会开始解析流图，并往集群提交

jobClient = executeAsync(jobName);
	Plan plan = createProgramPlan(jobName);
	xecutorFactory.getExecutor(configuration).execute(plan, configuration);
		AbstractSessionClusterExecutor.execute - 以session部署模式为例子
			jobGraph = ExecutorUtils.getJobGraph(pipeline, configuration);
			clusterClient = clusterClientProvider.getClusterClient();
			clusterClient.submitJob(jobGraph) - 向集群提交作业