flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-clients/src/main/java/org/apache/flink/client/CliFrontend.java
/**
	 * Sends a {@link org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint}
	 * message to the job manager.
	 */
Sends a  org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint
message to the job manager.
Sends a org . apache . flink . runtime . messages . JobManagerMessages . TriggerSavepoint message to the job manager .
private int triggerSavepoint(SavepointOptions options, JobID jobId, String savepointDirectory) {
		try {
			CustomCommandLine<?> activeCommandLine = getActiveCustomCommandLine(options.getCommandLine());
			ClusterClient client = activeCommandLine.retrieveCluster(options.getCommandLine(), config, configurationDirectory);
			try {
				logAndSysout("Triggering savepoint for job " + jobId + ".");
				CompletableFuture<String> savepointPathFuture = client.triggerSavepoint(jobId, savepointDirectory);
				String savepointPath;
				try {
					logAndSysout("Waiting for response...");
					savepointPath = savepointPathFuture.get();
				}
				catch (ExecutionException ee) {
					Throwable cause = ExceptionUtils.stripExecutionException(ee);
					throw new FlinkException("Triggering a savepoint for the job " + jobId + " failed.", cause);
				}
				logAndSysout("Savepoint completed. Path: " + savepointPath);
				logAndSysout("You can resume your program from this savepoint with the run command.");
				return 0;
			}
			finally {
				client.shutdown();
			}
		}
		catch (Throwable t) {
			return handleError(t);
		}
	}
private int triggerSavepoint ( SavepointOptions options , JobID jobId , String savepointDirectory ) { try { CustomCommandLine < ? > activeCommandLine = getActiveCustomCommandLine ( options . getCommandLine ( ) ) ; ClusterClient client = activeCommandLine . retrieveCluster ( options . getCommandLine ( ) , config , configurationDirectory ) ; try { logAndSysout ( "Triggering savepoint for job " + jobId + "." ) ; CompletableFuture < String > savepointPathFuture = client . triggerSavepoint ( jobId , savepointDirectory ) ; String savepointPath ; try { logAndSysout ( "Waiting for response..." ) ; savepointPath = savepointPathFuture . get ( ) ; } catch ( ExecutionException ee ) { Throwable cause = ExceptionUtils . stripExecutionException ( ee ) ; throw new FlinkException ( "Triggering a savepoint for the job " + jobId + " failed." , cause ) ; } logAndSysout ( "Savepoint completed. Path: " + savepointPath ) ; logAndSysout ( "You can resume your program from this savepoint with the run command." ) ; return 0 ; } finally { client . shutdown ( ) ; } } catch ( Throwable t ) { return handleError ( t ) ; } }
Sends a {@link org.apache.flink.runtime.messages.JobManagerMessages.TriggerSavepoint}
 message to the job manager.
Sends a { @link org . apache . flink . runtime . messages . JobManagerMessages . TriggerSavepoint } message to the job manager .