### Question:
CheckpointCoordinator { public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) { return triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic).isSuccess(); } 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 testStopPeriodicScheduler() throws Exception { final ExecutionAttemptID attemptID1 = new ExecutionAttemptID(); ExecutionVertex vertex1 = mockExecutionVertex(attemptID1); CheckpointCoordinator coord = new CheckpointCoordinator( new JobID(), 600000, 600000, 0, Integer.MAX_VALUE, CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new ExecutionVertex[] { vertex1 }, new StandaloneCheckpointIDCounter(), new StandaloneCompletedCheckpointStore(1), new MemoryStateBackend(), Executors.directExecutor(), SharedStateRegistry.DEFAULT_FACTORY); CheckpointTriggerResult triggerResult = coord.triggerCheckpoint( System.currentTimeMillis(), CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), null, true); assertTrue(triggerResult.isFailure()); assertEquals(CheckpointDeclineReason.PERIODIC_SCHEDULER_SHUTDOWN, triggerResult.getFailureReason()); triggerResult = coord.triggerCheckpoint( System.currentTimeMillis(), CheckpointProperties.forCheckpoint(CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION), null, false); assertFalse(triggerResult.isFailure()); }