flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
/**
	 * Creates a {@link ClusterClient} object from the given command line options and other parameters.
	 * @param options Command line options
	 * @param program The program for which to create the client.
	 * @throws Exception
	 */
Creates a  ClusterClient object from the given command line options and other parameters.
@param options Command line options
@param program The program for which to create the client.
@throws Exception
Creates a ClusterClient object from the given command line options and other parameters . @param options Command line options @param program The program for which to create the client . @throws Exception
protected ClusterClient createClient(
			CommandLineOptions options,
			PackagedProgram program) throws Exception {
		CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(options.getCommandLine());
		ClusterClient client;
		try {
			client = activeCommandLine.retrieveCluster(options.getCommandLine(), config, configurationDirectory);
			logAndSysout("Cluster configuration: " + client.getClusterIdentifier());
		} catch (UnsupportedOperationException e) {
			try {
				String applicationName = "Flink Application: " + program.getMainClassName();
				client = activeCommandLine.createCluster(
					applicationName,
					options.getCommandLine(),
					config,
					configurationDirectory,
					program.getAllLibraries());
				logAndSysout("Cluster started: " + client.getClusterIdentifier());
			} catch (UnsupportedOperationException e2) {
				throw new IllegalConfigurationException(
					"The JobManager address is neither provided at the command-line, " +
						"nor configured in flink-conf.yaml.");
			}
		}
		final InetSocketAddress jobManagerAddress = client.getJobManagerAddress();
		logAndSysout("Using address " + jobManagerAddress.getHostString() + ":" + jobManagerAddress.getPort() + " to connect to JobManager.");
		try {
			logAndSysout("JobManager web interface address " + client.getWebInterfaceURL());
		} catch (UnsupportedOperationException uoe) {
			logAndSysout("JobManager web interface not active.");
		}
		return client;
	}
protected ClusterClient createClient ( CommandLineOptions options , PackagedProgram program ) throws Exception { CustomCommandLine < ? > activeCommandLine = getActiveCustomCommandLine ( options . getCommandLine ( ) ) ; ClusterClient client ; try { client = activeCommandLine . retrieveCluster ( options . getCommandLine ( ) , config , configurationDirectory ) ; logAndSysout ( "Cluster configuration: " + client . getClusterIdentifier ( ) ) ; } catch ( UnsupportedOperationException e ) { try { String applicationName = "Flink Application: " + program . getMainClassName ( ) ; client = activeCommandLine . createCluster ( applicationName , options . getCommandLine ( ) , config , configurationDirectory , program . getAllLibraries ( ) ) ; logAndSysout ( "Cluster started: " + client . getClusterIdentifier ( ) ) ; } catch ( UnsupportedOperationException e2 ) { throw new IllegalConfigurationException ( "The JobManager address is neither provided at the command-line, " + "nor configured in flink-conf.yaml." ) ; } } final InetSocketAddress jobManagerAddress = client . getJobManagerAddress ( ) ; logAndSysout ( "Using address " + jobManagerAddress . getHostString ( ) + ":" + jobManagerAddress . getPort ( ) + " to connect to JobManager." ) ; try { logAndSysout ( "JobManager web interface address " + client . getWebInterfaceURL ( ) ) ; } catch ( UnsupportedOperationException uoe ) { logAndSysout ( "JobManager web interface not active." ) ; } return client ; }
Creates a {@link ClusterClient} object from the given command line options and other parameters.
Creates a { @link ClusterClient } object from the given command line options and other parameters .
// Get the custom command-line (e.g. Standalone/Yarn/Mesos) // Avoid resolving the JobManager Gateway here to prevent blocking until we invoke the user's program.