flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
/**
	 * Triggers a savepoint with the given savepoint directory as a target.
	 *
	 * @param timestamp The timestamp for the savepoint.
	 * @param targetDirectory Target directory for the savepoint.
	 * @return A future to the completed checkpoint
	 * @throws IllegalStateException If no savepoint directory has been
	 *                               specified and no default savepoint directory has been
	 *                               configured
	 * @throws Exception             Failures during triggering are forwarded
	 */
Triggers a savepoint with the given savepoint directory as a target.
@param timestamp The timestamp for the savepoint.
@param targetDirectory Target directory for the savepoint.
@return A future to the completed checkpoint
@throws IllegalStateException If no savepoint directory has been
specified and no default savepoint directory has been
configured
@throws Exception             Failures during triggering are forwarded
Triggers a savepoint with the given savepoint directory as a target . @param timestamp The timestamp for the savepoint . @param targetDirectory Target directory for the savepoint . @return A future to the completed checkpoint @throws IllegalStateException If no savepoint directory has been specified and no default savepoint directory has been configured @throws Exception Failures during triggering are forwarded
public CompletableFuture<CompletedCheckpoint> triggerSavepoint(long timestamp, String targetDirectory) throws Exception {
		checkNotNull(targetDirectory, "Savepoint target directory");
		CheckpointProperties props = CheckpointProperties.forStandardSavepoint();
		final String savepointDirectory = SavepointStore
			.createSavepointDirectory(targetDirectory, job);
		CheckpointTriggerResult triggerResult = triggerCheckpoint(
			timestamp,
			props,
			savepointDirectory,
			false);
		CompletableFuture<CompletedCheckpoint> result;
		if (triggerResult.isSuccess()) {
			result = triggerResult.getPendingCheckpoint().getCompletionFuture();
		} else {
			Throwable cause = new Exception("Failed to trigger savepoint: " + triggerResult.getFailureReason().message());
			return FutureUtils.completedExceptionally(cause);
		}
		result.whenCompleteAsync(
			(CompletedCheckpoint checkpoint, Throwable throwable) -> {
				if (throwable != null) {
					try {
						SavepointStore.deleteSavepointDirectory(savepointDirectory);
					} catch (Throwable t) {
						LOG.warn("Failed to delete savepoint directory " + savepointDirectory
							+ " after failed savepoint.", t);
					}
				}
			},
			executor);
		return result;
	}
public CompletableFuture < CompletedCheckpoint > triggerSavepoint ( long timestamp , String targetDirectory ) throws Exception { checkNotNull ( targetDirectory , "Savepoint target directory" ) ; CheckpointProperties props = CheckpointProperties . forStandardSavepoint ( ) ; final String savepointDirectory = SavepointStore . createSavepointDirectory ( targetDirectory , job ) ; CheckpointTriggerResult triggerResult = triggerCheckpoint ( timestamp , props , savepointDirectory , false ) ; CompletableFuture < CompletedCheckpoint > result ; if ( triggerResult . isSuccess ( ) ) { result = triggerResult . getPendingCheckpoint ( ) . getCompletionFuture ( ) ; } else { Throwable cause = new Exception ( "Failed to trigger savepoint: " + triggerResult . getFailureReason ( ) . message ( ) ) ; return FutureUtils . completedExceptionally ( cause ) ; } result . whenCompleteAsync ( ( CompletedCheckpoint checkpoint , Throwable throwable ) -> { if ( throwable != null ) { try { SavepointStore . deleteSavepointDirectory ( savepointDirectory ) ; } catch ( Throwable t ) { LOG . warn ( "Failed to delete savepoint directory " + savepointDirectory + " after failed savepoint." , t ) ; } } } , executor ) ; return result ; }
Triggers a savepoint with the given savepoint directory as a target.
Triggers a savepoint with the given savepoint directory as a target .
// Create the unique savepoint directory // Make sure to remove the created base directory on Exceptions