CheckpointCoordinator { 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; } CheckpointCoordinator(
			JobID job,
			long baseInterval,
			long checkpointTimeout,
			long minPauseBetweenCheckpoints,
			int maxConcurrentCheckpointAttempts,
			ExternalizedCheckpointSettings externalizeSettings,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			@Nullable String checkpointDirectory,
			Executor executor,
			SharedStateRegistryFactory sharedStateRegistryFactory); boolean addMasterHook(MasterTriggerRestoreHook<?> hook); int getNumberOfRegisteredMasterHooks(); void setCheckpointStatsTracker(@Nullable CheckpointStatsTracker statsTracker); void shutdown(JobStatus jobStatus); boolean isShutdown(); CompletableFuture<CompletedCheckpoint> triggerSavepoint(long timestamp, String targetDirectory); boolean triggerCheckpoint(long timestamp, boolean isPeriodic); @VisibleForTesting @Internal CompletableFuture<CompletedCheckpoint> triggerCheckpoint(long timestamp, CheckpointOptions options); void receiveDeclineMessage(DeclineCheckpoint message); boolean receiveAcknowledgeMessage(AcknowledgeCheckpoint message); void failUnacknowledgedPendingCheckpointsFor(ExecutionAttemptID executionAttemptId, Throwable cause); boolean restoreLatestCheckpointedState(
			Map<JobVertexID, ExecutionJobVertex> tasks,
			boolean errorIfNoCheckpoint,
			boolean allowNonRestoredState); boolean restoreSavepoint(
			String savepointPath, 
			boolean allowNonRestored,
			Map<JobVertexID, ExecutionJobVertex> tasks,
			ClassLoader userClassLoader); int getNumberOfPendingCheckpoints(); int getNumberOfRetainedSuccessfulCheckpoints(); Map<Long, PendingCheckpoint> getPendingCheckpoints(); List<CompletedCheckpoint> getSuccessfulCheckpoints(); CompletedCheckpointStore getCheckpointStore(); CheckpointIDCounter getCheckpointIdCounter(); long getCheckpointTimeout(); boolean isPeriodicCheckpointingConfigured(); void startCheckpointScheduler(); void stopCheckpointScheduler(); JobStatusListener createActivatorDeactivator();  }
@Test public void testMinDelayBetweenSavepoints() throws Exception { JobID jobId = new JobID(); final ExecutionAttemptID attemptID1 = new ExecutionAttemptID(); ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); CheckpointCoordinator coord = new CheckpointCoordinator( jobId, 100000, 200000, 100000000L, 1, ExternalizedCheckpointSettings.none(), new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new StandaloneCheckpointIDCounter(), new StandaloneCompletedCheckpointStore(2), null, Executors.directExecutor(), SharedStateRegistry.DEFAULT_FACTORY); String savepointDir = tmpFolder.newFolder().getAbsolutePath(); CompletableFuture<CompletedCheckpoint> savepoint0 = coord.triggerSavepoint(0, savepointDir); assertFalse("Did not trigger savepoint", savepoint0.isDone()); CompletableFuture<CompletedCheckpoint> savepoint1 = coord.triggerSavepoint(1, savepointDir); assertFalse("Did not trigger savepoint", savepoint1.isDone()); }