flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-yarn/src/main/java/org/apache/flink/yarn/YarnClusterClient.java
/**
	 * Shuts down the Yarn application.
	 */
Shuts down the Yarn application.
Shuts down the Yarn application .
public void shutdownCluster() {
		if (hasBeenShutDown.getAndSet(true)) {
			return;
		}
		if (!isConnected) {
			throw new IllegalStateException("The cluster has been not been connected to the ApplicationMaster.");
		}
		try {
			Runtime.getRuntime().removeShutdownHook(clientShutdownHook);
		} catch (IllegalStateException e) {
		}
		LOG.info("Sending shutdown request to the Application Master");
		try {
			Future<Object> response =
				Patterns.ask(applicationClient.get(),
					new YarnMessages.LocalStopYarnSession(getApplicationStatus(),
							"Flink YARN Client requested shutdown"),
					new Timeout(akkaDuration));
			Await.ready(response, akkaDuration);
		} catch (Exception e) {
			LOG.warn("Error while stopping YARN cluster.", e);
		}
		try {
			File propertiesFile = FlinkYarnSessionCli.getYarnPropertiesLocation(flinkConfig);
			if (propertiesFile.isFile()) {
				if (propertiesFile.delete()) {
					LOG.info("Deleted Yarn properties file at {}", propertiesFile.getAbsoluteFile().toString());
				} else {
					LOG.warn("Couldn't delete Yarn properties file at {}", propertiesFile.getAbsoluteFile().toString());
				}
			}
		} catch (Exception e) {
			LOG.warn("Exception while deleting the JobManager address file", e);
		}
		try {
			pollingRunner.stopRunner();
			pollingRunner.join(1000);
		} catch (InterruptedException e) {
			LOG.warn("Shutdown of the polling runner was interrupted", e);
			Thread.currentThread().interrupt();
		}
		try {
			ApplicationReport appReport = yarnClient.getApplicationReport(appId);
			LOG.info("Application " + appId + " finished with state " + appReport
				.getYarnApplicationState() + " and final state " + appReport
				.getFinalApplicationStatus() + " at " + appReport.getFinishTime());
			if (appReport.getYarnApplicationState() == YarnApplicationState.FAILED || appReport.getYarnApplicationState()
				== YarnApplicationState.KILLED) {
				LOG.warn("Application failed. Diagnostics " + appReport.getDiagnostics());
				LOG.warn("If log aggregation is activated in the Hadoop cluster, we recommend to retrieve "
					+ "the full application log using this command:"
					+ System.lineSeparator()
					+ "\tyarn logs -applicationId " + appReport.getApplicationId()
					+ System.lineSeparator()
					+ "(It sometimes takes a few seconds until the logs are aggregated)");
			}
		} catch (Exception e) {
			LOG.warn("Couldn't get final report", e);
		}
		LOG.info("YARN Client is shutting down");
		yarnClient.stop(); 
		yarnClient = null; 
	}
public void shutdownCluster ( ) { if ( hasBeenShutDown . getAndSet ( true ) ) { return ; } if ( ! isConnected ) { throw new IllegalStateException ( "The cluster has been not been connected to the ApplicationMaster." ) ; } try { Runtime . getRuntime ( ) . removeShutdownHook ( clientShutdownHook ) ; } catch ( IllegalStateException e ) { } LOG . info ( "Sending shutdown request to the Application Master" ) ; try { Future < Object > response = Patterns . ask ( applicationClient . get ( ) , new YarnMessages . LocalStopYarnSession ( getApplicationStatus ( ) , "Flink YARN Client requested shutdown" ) , new Timeout ( akkaDuration ) ) ; Await . ready ( response , akkaDuration ) ; } catch ( Exception e ) { LOG . warn ( "Error while stopping YARN cluster." , e ) ; } try { File propertiesFile = FlinkYarnSessionCli . getYarnPropertiesLocation ( flinkConfig ) ; if ( propertiesFile . isFile ( ) ) { if ( propertiesFile . delete ( ) ) { LOG . info ( "Deleted Yarn properties file at {}" , propertiesFile . getAbsoluteFile ( ) . toString ( ) ) ; } else { LOG . warn ( "Couldn't delete Yarn properties file at {}" , propertiesFile . getAbsoluteFile ( ) . toString ( ) ) ; } } } catch ( Exception e ) { LOG . warn ( "Exception while deleting the JobManager address file" , e ) ; } try { pollingRunner . stopRunner ( ) ; pollingRunner . join ( 1000 ) ; } catch ( InterruptedException e ) { LOG . warn ( "Shutdown of the polling runner was interrupted" , e ) ; Thread . currentThread ( ) . interrupt ( ) ; } try { ApplicationReport appReport = yarnClient . getApplicationReport ( appId ) ; LOG . info ( "Application " + appId + " finished with state " + appReport . getYarnApplicationState ( ) + " and final state " + appReport . getFinalApplicationStatus ( ) + " at " + appReport . getFinishTime ( ) ) ; if ( appReport . getYarnApplicationState ( ) == YarnApplicationState . FAILED || appReport . getYarnApplicationState ( ) == YarnApplicationState . KILLED ) { LOG . warn ( "Application failed. Diagnostics " + appReport . getDiagnostics ( ) ) ; LOG . warn ( "If log aggregation is activated in the Hadoop cluster, we recommend to retrieve " + "the full application log using this command:" + System . lineSeparator ( ) + "\tyarn logs -applicationId " + appReport . getApplicationId ( ) + System . lineSeparator ( ) + "(It sometimes takes a few seconds until the logs are aggregated)" ) ; } } catch ( Exception e ) { LOG . warn ( "Couldn't get final report" , e ) ; } LOG . info ( "YARN Client is shutting down" ) ; yarnClient . stop ( ) ; yarnClient = null ; }
Shuts down the Yarn application.
Shuts down the Yarn application .
// we are already in the shutdown hook // actorRunner is using the yarnClient. // set null to clearly see if somebody wants to access it afterwards.