### Question:
CheckpointCoordinator { public CompletableFuture<CompletedCheckpoint> triggerSavepoint( long timestamp, @Nullable String targetLocation) { CheckpointProperties props = CheckpointProperties.forSavepoint(); CheckpointTriggerResult triggerResult = triggerCheckpoint( timestamp, props, targetLocation, false); if (triggerResult.isSuccess()) { return triggerResult.getPendingCheckpoint().getCompletionFuture(); } else { Throwable cause = new CheckpointTriggerException("Failed to trigger savepoint.", triggerResult.getFailureReason()); return FutureUtils.completedExceptionally(cause); } } CheckpointCoordinator(
			JobID job,
			long baseInterval,
			long checkpointTimeout,
			long minPauseBetweenCheckpoints,
			int maxConcurrentCheckpointAttempts,
			CheckpointRetentionPolicy retentionPolicy,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			CheckpointIDCounter checkpointIDCounter,
			CompletedCheckpointStore completedCheckpointStore,
			StateBackend checkpointStateBackend,
			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,
			@Nullable String targetLocation); boolean triggerCheckpoint(long timestamp, boolean isPeriodic); @VisibleForTesting CheckpointTriggerResult triggerCheckpoint(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic); 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 savepointPointer,
			boolean allowNonRestored,
			Map<JobVertexID, ExecutionJobVertex> tasks,
			ClassLoader userClassLoader); int getNumberOfPendingCheckpoints(); int getNumberOfRetainedSuccessfulCheckpoints(); Map<Long, PendingCheckpoint> getPendingCheckpoints(); List<CompletedCheckpoint> getSuccessfulCheckpoints(); CheckpointStorage getCheckpointStorage(); CompletedCheckpointStore getCheckpointStore(); CheckpointIDCounter getCheckpointIdCounter(); long getCheckpointTimeout(); boolean isPeriodicCheckpointingConfigured(); void startCheckpointScheduler(); void stopCheckpointScheduler(); JobStatusListener createActivatorDeactivator();  }### Answer:
@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, CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new StandaloneCheckpointIDCounter(), new StandaloneCompletedCheckpointStore(2), new MemoryStateBackend(), 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()); }