Flink 启动流程之 yarn-session

// org.apache.flink.yarn.cli.FlinkYarnSessionCli
public static void main(final String[] args) {
  // flink conf 目录载入配置:ip,port,jmHeap,tmHeap,solts,parllelism ...
  final String configurationDirectory = CliFrontend.getConfigurationDirectoryFromEnv();
  final Configuration flinkConfiguration = GlobalConfiguration.loadConfiguration();

  int retCode;

  try {
    final FlinkYarnSessionCli cli = new FlinkYarnSessionCli(
      flinkConfiguration,
      configurationDirectory,
      "",
      ""); // no prefix for the YARN session

    SecurityUtils.install(new SecurityConfiguration(flinkConfiguration));
    retCode = SecurityUtils.getInstalledContext().runSecured(() -> cli.run(args));
   ...
  System.exit(retCode);
}

// org.apache.flink.yarn.cli.FlinkYarnSessionCli
public int run(String[] args) throws CliArgsException, FlinkException {
  final CommandLine cmd = parseCommandLineOptions(args, true);
  ...
  // 设置资源参数
  final Configuration configuration = applyCommandLineOptionsToConfiguration(cmd);
  final ClusterClientFactory<ApplicationId> yarnClusterClientFactory = clusterClientServiceLoader.getClusterClientFactory(configuration);
  // YarnClusterDescriptor 很重要，Flink On Yarn 的信息
  final YarnClusterDescriptor yarnClusterDescriptor = (YarnClusterDescriptor) yarnClusterClientFactory.createClusterDescriptor(configuration);

  try {
    // 可以查询集群，输出Memory，vCores，HealthReport
    if (cmd.hasOption(query.getOpt())) {
      final String description = yarnClusterDescriptor.getClusterDescription();
      System.out.println(description);
      return 0;
    } else {
      final ClusterClientProvider<ApplicationId> clusterClientProvider;
      final ApplicationId yarnApplicationId;
      
      if (cmd.hasOption(applicationId.getOpt())) {
        // 设置固定 appid，并获取 clusterClient
        yarnApplicationId = ConverterUtils.toApplicationId(cmd.getOptionValue(applicationId.getOpt()));
        clusterClientProvider = yarnClusterDescriptor.retrieve(yarnApplicationId);
      } else {
        final ClusterSpecification clusterSpecification = yarnClusterClientFactory.getClusterSpecification(configuration);

        clusterClientProvider = yarnClusterDescriptor.deploySessionCluster(clusterSpecification);
        // ClusterClient：封装了将程序提交到远程集群所需的功能，很重要
        ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();
        //------------------ ClusterClient deployed, handle connection details
        yarnApplicationId = clusterClient.getClusterId();
        // web UI
        try {
          System.out.println("JobManager Web Interface: " + clusterClient.getWebInterfaceURL());
         // yarnApplicationId 记录到本地文件，一般为 /tmp/.yarn-properties-user
         // contnt: applicationID=application_1593474472940_0001
          // flink run 启动时会根据此 id 来提交
          writeYarnPropertiesFile(
            yarnApplicationId,
            dynamicPropertiesEncoded);
        } 
      ...
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

        final YarnApplicationStatusMonitor yarnApplicationStatusMonitor = new YarnApplicationStatusMonitor(
          yarnClusterDescriptor.getYarnClient(),
          yarnApplicationId,
          new ScheduledExecutorServiceAdapter(scheduledExecutorService));
        Thread shutdownHook = ShutdownHookUtil.addShutdownHook(
          () -> shutdownCluster(
            clusterClientProvider.getClusterClient(),
            scheduledExecutorService,
            yarnApplicationStatusMonitor),
          getClass().getSimpleName(),
          LOG);
        try {
          runInteractiveCli(
            yarnApplicationStatusMonitor,
            acceptInteractiveInput);
        } 
      ...
  return 0;
}

clusterClientServiceLoader = DefaultClusterClientServiceLoader
  final ClusterClientFactory<ApplicationId> yarnClusterClientFactory = clusterClientServiceLoader.getClusterClientFactory(configuration);

final YarnClusterDescriptor yarnClusterDescriptor = (YarnClusterDescriptor) yarnClusterClientFactory.createClusterDescriptor(configuration);

// org.apache.flink.client.deployment.DefaultClusterClientServiceLoader
// 加载所有 ClusterClientFactory 类
private static final ServiceLoader<ClusterClientFactory> defaultLoader = ServiceLoader.load(ClusterClientFactory.class);
	@Override
	public <ClusterID> ClusterClientFactory<ClusterID> getClusterClientFactory(final Configuration configuration) {
		checkNotNull(configuration);

	final List<ClusterClientFactory> compatibleFactories = new ArrayList<>();
	final Iterator<ClusterClientFactory> factories = defaultLoader.iterator();
		while (factories.hasNext()) {
			try {
				final ClusterClientFactory factory = factories.next();
        // 比对当前的 ClusterClientFactory name 和configuration 中 target 是否相同
				if (factory != null && factory.isCompatibleWith(configuration)) {
					compatibleFactories.add(factory);
				}
			} catch (Throwable e) {
				if (e.getCause() instanceof NoClassDefFoundError) {
					LOG.info("Could not load factory due to missing dependencies.");
				} else {
					throw e;
				}
			}
		}

		if (compatibleFactories.size() > 1) {
			...
			throw new IllegalStateException("Multiple compatible client factories found for:\n" + String.join("\n", configStr) + ".");
		}
		return compatibleFactories.isEmpty() ? null : (ClusterClientFactory<ClusterID>) compatibleFactories.get(0);
	}
// org.apache.flink.yarn.YarnClusterClientFactory
@Override
public YarnClusterDescriptor createClusterDescriptor(Configuration configuration) {
  checkNotNull(configuration);
  return getClusterDescriptor(configuration);
}
private YarnClusterDescriptor getClusterDescriptor(Configuration configuration) {
  final YarnClient yarnClient = YarnClient.createYarnClient();
  final YarnConfiguration yarnConfiguration = new YarnConfiguration();
  yarnClient.init(yarnConfiguration);
  yarnClient.start();
  return new YarnClusterDescriptor(
    configuration,
    yarnConfiguration,
    yarnClient,
    YarnClientYarnClusterInformationRetriever.create(yarnClient),
    false);
}

// 获取 flink config.masterMemoryMB,taskManagerMemoryMB,slotsPerTaskManager
final ClusterSpecification clusterSpecification = yarnClusterClientFactory.getClusterSpecification(configuration);
clusterClientProvider = yarnClusterDescriptor.deploySessionCluster(clusterSpecification);
ClusterClient<ApplicationId> clusterClient = clusterClientProvider.getClusterClient();

// org.apache.flink.yarn.YarnClusterDescriptor
public ClusterClientProvider<ApplicationId> deploySessionCluster(ClusterSpecification clusterSpecification) throws ClusterDeploymentException {
  try {
    return deployInternal(
      clusterSpecification,
      "Flink session cluster",
      getYarnSessionClusterEntrypoint(),
      null,
      false);
  } catch (Exception e) {
    throw new ClusterDeploymentException("Couldn't deploy Yarn session cluster", e);
  }
}
private ClusterClientProvider<ApplicationId> deployInternal(
			ClusterSpecification clusterSpecification,
			String applicationName,
			String yarnClusterEntrypoint,
			@Nullable JobGraph jobGraph,
			boolean detached) throws Exception {

		if (UserGroupInformation.isSecurityEnabled()) {
			// note: UGI::hasKerberosCredentials inaccurately reports false
			// for logins based on a keytab (fixed in Hadoop 2.6.1, see HADOOP-10786),
			// so we check only in ticket cache scenario.
			boolean useTicketCache = flinkConfiguration.getBoolean(SecurityOptions.KERBEROS_LOGIN_USETICKETCACHE);

			UserGroupInformation loginUser = UserGroupInformation.getCurrentUser();
			if (loginUser.getAuthenticationMethod() == UserGroupInformation.AuthenticationMethod.KERBEROS
					&& useTicketCache && !loginUser.hasKerberosCredentials()) {
				LOG.error("Hadoop security with Kerberos is enabled but the login user does not have Kerberos credentials");
				throw new RuntimeException("Hadoop security with Kerberos is enabled but the login user " +
						"does not have Kerberos credentials");
			}
		}
		isReadyForDeployment(clusterSpecification);
		// ------------------ Check if the specified queue exists --------------------
		checkYarnQueues(yarnClient);
		// ------------------ Check if the YARN ClusterClient has the requested resources --------------
		// Create application via yarnClient
		final YarnClientApplication yarnApplication = yarnClient.createApplication();
		final GetNewApplicationResponse appResponse = yarnApplication.getNewApplicationResponse();

		Resource maxRes = appResponse.getMaximumResourceCapability();

		final ClusterResourceDescription freeClusterMem;
		try {
      // 获取集群空闲资源
			freeClusterMem = getCurrentFreeClusterResources(yarnClient);
		} catch (YarnException | IOException e) {
			failSessionDuringDeployment(yarnClient, yarnApplication);
			throw new YarnDeploymentException("Could not retrieve information about free cluster resources.", e);
		}
		final int yarnMinAllocationMB = yarnConfiguration.getInt(YarnConfiguration.RM_SCHEDULER_MINIMUM_ALLOCATION_MB, 0);

		final ClusterSpecification validClusterSpecification;
		try {
      // 检查资源是否满足
			validClusterSpecification = validateClusterResources(
					clusterSpecification,
					yarnMinAllocationMB,
					maxRes,
					freeClusterMem);
		} catch (YarnDeploymentException yde) {
			failSessionDuringDeployment(yarnClient, yarnApplication);
			throw yde;
		}
		LOG.info("Cluster specification: {}", validClusterSpecification);
		final ClusterEntrypoint.ExecutionMode executionMode = detached ?
				ClusterEntrypoint.ExecutionMode.DETACHED
				: ClusterEntrypoint.ExecutionMode.NORMAL;
		flinkConfiguration.setString(ClusterEntrypoint.EXECUTION_MODE, executionMode.toString());
   // 通过 YARN 将封装好的 ApplicationSubmissionContext 提交到 yarn，返回 report
   // yarnClusterEntrypoint 就是yarn 集群入口点，类似之前分析的 StandaloneSessionClusterEntrypoint
		ApplicationReport report = startAppMaster(
				flinkConfiguration,
				applicationName,
				yarnClusterEntrypoint,
				jobGraph,
				yarnClient,
				yarnApplication,
				validClusterSpecification);
		// print the application id for user to cancel themselves.
		if (detached) {
			final ApplicationId yarnApplicationId = report.getApplicationId();
			logDetachedClusterInformation(yarnApplicationId, LOG);
		}
		setClusterEntrypointInfoToConfig(report);
		return () -> {
			try {
				return new RestClusterClient<>(flinkConfiguration, report.getApplicationId());
			} catch (Exception e) {
				throw new RuntimeException("Error while creating RestClusterClient.", e);
			}
		};
	}

/**
 * A {@link ClusterClient} implementation that communicates via HTTP REST requests.
 * org.apache.flink.client.program.rest.RestClusterClient
 */