flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/RemoteStreamEnvironment.java
/**
	 * Executes the remote job.
	 *
	 * @param streamGraph
	 *            Stream Graph to execute
	 * @param jarFiles
	 * 			  List of jar file URLs to ship to the cluster
	 * @return The result of the job execution, containing elapsed time and accumulators.
	 */
Executes the remote job.
@param streamGraph
Stream Graph to execute
@param jarFiles
List of jar file URLs to ship to the cluster
@return The result of the job execution, containing elapsed time and accumulators.
Executes the remote job . @param streamGraph Stream Graph to execute @param jarFiles List of jar file URLs to ship to the cluster @return The result of the job execution containing elapsed time and accumulators .
protected JobExecutionResult executeRemotely(StreamGraph streamGraph, List<URL> jarFiles) throws ProgramInvocationException {
		if (LOG.isInfoEnabled()) {
			LOG.info("Running remotely at {}:{}", host, port);
		}
		ClassLoader usercodeClassLoader = JobWithJars.buildUserCodeClassLoader(jarFiles, globalClasspaths,
			getClass().getClassLoader());
		Configuration configuration = new Configuration();
		configuration.addAll(this.clientConfiguration);
		configuration.setString(JobManagerOptions.ADDRESS, host);
		configuration.setInteger(JobManagerOptions.PORT, port);
		ClusterClient client;
		try {
			client = new StandaloneClusterClient(configuration);
			client.setPrintStatusDuringExecution(getConfig().isSysoutLoggingEnabled());
		}
		catch (Exception e) {
			throw new ProgramInvocationException("Cannot establish connection to JobManager: " + e.getMessage(), e);
		}
		try {
			return client.run(streamGraph, jarFiles, globalClasspaths, usercodeClassLoader).getJobExecutionResult();
		}
		catch (ProgramInvocationException e) {
			throw e;
		}
		catch (Exception e) {
			String term = e.getMessage() == null ? "." : (": " + e.getMessage());
			throw new ProgramInvocationException("The program execution failed" + term, e);
		}
		finally {
			try {
				client.shutdown();
			} catch (Exception e) {
				LOG.warn("Could not properly shut down the cluster client.", e);
			}
		}
	}
protected JobExecutionResult executeRemotely ( StreamGraph streamGraph , List < URL > jarFiles ) throws ProgramInvocationException { if ( LOG . isInfoEnabled ( ) ) { LOG . info ( "Running remotely at {}:{}" , host , port ) ; } ClassLoader usercodeClassLoader = JobWithJars . buildUserCodeClassLoader ( jarFiles , globalClasspaths , getClass ( ) . getClassLoader ( ) ) ; Configuration configuration = new Configuration ( ) ; configuration . addAll ( this . clientConfiguration ) ; configuration . setString ( JobManagerOptions . ADDRESS , host ) ; configuration . setInteger ( JobManagerOptions . PORT , port ) ; ClusterClient client ; try { client = new StandaloneClusterClient ( configuration ) ; client . setPrintStatusDuringExecution ( getConfig ( ) . isSysoutLoggingEnabled ( ) ) ; } catch ( Exception e ) { throw new ProgramInvocationException ( "Cannot establish connection to JobManager: " + e . getMessage ( ) , e ) ; } try { return client . run ( streamGraph , jarFiles , globalClasspaths , usercodeClassLoader ) . getJobExecutionResult ( ) ; } catch ( ProgramInvocationException e ) { throw e ; } catch ( Exception e ) { String term = e . getMessage ( ) == null ? "." : ( ": " + e . getMessage ( ) ) ; throw new ProgramInvocationException ( "The program execution failed" + term , e ) ; } finally { try { client . shutdown ( ) ; } catch ( Exception e ) { LOG . warn ( "Could not properly shut down the cluster client." , e ) ; } } }
Executes the remote job.
Executes the remote job .