flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
/**
	 * Executions the run action.
	 *
	 * @param args Command line arguments for the run action.
	 */
Executions the run action.
@param args Command line arguments for the run action.
Executions the run action . @param args Command line arguments for the run action .
protected int run(String[] args) {
		LOG.info("Running 'run' command.");
		RunOptions options;
		try {
			options = CliFrontendParser.parseRunCommand(args);
		}
		catch (CliArgsException e) {
			return handleArgException(e);
		}
		catch (Throwable t) {
			return handleError(t);
		}
		if (options.isPrintHelp()) {
			CliFrontendParser.printHelpForRun();
			return 0;
		}
		if (options.getJarFilePath() == null) {
			return handleArgException(new CliArgsException("The program JAR file was not specified."));
		}
		PackagedProgram program;
		try {
			LOG.info("Building program from JAR file");
			program = buildProgram(options);
		}
		catch (FileNotFoundException e) {
			return handleArgException(e);
		}
		catch (Throwable t) {
			return handleError(t);
		}
		ClusterClient client = null;
		try {
			client = createClient(options, program);
			client.setPrintStatusDuringExecution(options.getStdoutLogging());
			client.setDetached(options.getDetachedMode());
			LOG.debug("Client slots is set to {}", client.getMaxSlots());
			LOG.debug(options.getSavepointRestoreSettings().toString());
			int userParallelism = options.getParallelism();
			LOG.debug("User parallelism is set to {}", userParallelism);
			if (client.getMaxSlots() != -1 && userParallelism == -1) {
				logAndSysout("Using the parallelism provided by the remote cluster ("
					+ client.getMaxSlots() + "). "
					+ "To use another parallelism, set it at the ./bin/flink client.");
				userParallelism = client.getMaxSlots();
			} else if (ExecutionConfig.PARALLELISM_DEFAULT == userParallelism) {
				userParallelism = defaultParallelism;
			}
			return executeProgram(program, client, userParallelism);
		}
		catch (Throwable t) {
			return handleError(t);
		}
		finally {
			if (client != null) {
				try {
					client.shutdown();
				} catch (Exception e) {
					LOG.warn("Could not properly shut down the cluster client.", e);
				}
			}
			if (program != null) {
				program.deleteExtractedLibraries();
			}
		}
	}
protected int run ( String [ ] args ) { LOG . info ( "Running 'run' command." ) ; RunOptions options ; try { options = CliFrontendParser . parseRunCommand ( args ) ; } catch ( CliArgsException e ) { return handleArgException ( e ) ; } catch ( Throwable t ) { return handleError ( t ) ; } if ( options . isPrintHelp ( ) ) { CliFrontendParser . printHelpForRun ( ) ; return 0 ; } if ( options . getJarFilePath ( ) == null ) { return handleArgException ( new CliArgsException ( "The program JAR file was not specified." ) ) ; } PackagedProgram program ; try { LOG . info ( "Building program from JAR file" ) ; program = buildProgram ( options ) ; } catch ( FileNotFoundException e ) { return handleArgException ( e ) ; } catch ( Throwable t ) { return handleError ( t ) ; } ClusterClient client = null ; try { client = createClient ( options , program ) ; client . setPrintStatusDuringExecution ( options . getStdoutLogging ( ) ) ; client . setDetached ( options . getDetachedMode ( ) ) ; LOG . debug ( "Client slots is set to {}" , client . getMaxSlots ( ) ) ; LOG . debug ( options . getSavepointRestoreSettings ( ) . toString ( ) ) ; int userParallelism = options . getParallelism ( ) ; LOG . debug ( "User parallelism is set to {}" , userParallelism ) ; if ( client . getMaxSlots ( ) != - 1 && userParallelism == - 1 ) { logAndSysout ( "Using the parallelism provided by the remote cluster (" + client . getMaxSlots ( ) + "). " + "To use another parallelism, set it at the ./bin/flink client." ) ; userParallelism = client . getMaxSlots ( ) ; } else if ( ExecutionConfig . PARALLELISM_DEFAULT == userParallelism ) { userParallelism = defaultParallelism ; } return executeProgram ( program , client , userParallelism ) ; } catch ( Throwable t ) { return handleError ( t ) ; } finally { if ( client != null ) { try { client . shutdown ( ) ; } catch ( Exception e ) { LOG . warn ( "Could not properly shut down the cluster client." , e ) ; } } if ( program != null ) { program . deleteExtractedLibraries ( ) ; } } }
Executions the run action.
Executions the run action .
// evaluate help flag