@Test
public void testStopPeriodicScheduler() throws Exception {

	// create some mock Execution vertices that receive the checkpoint trigger messages
	final ExecutionAttemptID attemptID1 = new ExecutionAttemptID();
	ExecutionVertex vertex1 = mockExecutionVertex(attemptID1);
	// set up the coordinator and validate the initial state
	CheckpointCoordinator coord = new CheckpointCoordinator(
		new JobID(),
		600000,
		600000,
		0,
		Integer.MAX_VALUE,
		ExternalizedCheckpointSettings.none(),
		new ExecutionVertex[] { vertex1 },
		new ExecutionVertex[] { vertex1 },
		new ExecutionVertex[] { vertex1 },
		new StandaloneCheckpointIDCounter(),
		new StandaloneCompletedCheckpointStore(1),
		null,
		Executors.directExecutor(),
		SharedStateRegistry.DEFAULT_FACTORY);
	// Periodic
	CheckpointTriggerResult triggerResult = coord.triggerCheckpoint(
			System.currentTimeMillis(),
			CheckpointProperties.forStandardCheckpoint(),
			null,
			true);
	assertTrue(triggerResult.isFailure());
	assertEquals(CheckpointDeclineReason.PERIODIC_SCHEDULER_SHUTDOWN, triggerResult.getFailureReason());
	// Not periodic
	triggerResult = coord.triggerCheckpoint(
			System.currentTimeMillis(),
			CheckpointProperties.forStandardCheckpoint(),
			null,
			false);
	assertFalse(triggerResult.isFailure());
}