@Test
public void testShutdownDoesNotCleanUpCompletedCheckpointsWithFileSystemStore() throws Exception {

	JobID jobId = new JobID();
	long checkpointTimeout = 60 * 1000;
	long timestamp = 1272635;
	ExecutionVertex[] vertices = new ExecutionVertex[] {
			mockExecutionVertex(jobId),
			mockExecutionVertex(jobId) };
	MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
	// Temporary directory for file state backend
	final File tmpDir = CommonTestUtils.createTempDirectory();
	try {
		FileSystemStateStore<CompletedCheckpoint> savepointStore = new FileSystemStateStore<>(
				tmpDir.toURI().toString(), "sp-");
		SavepointCoordinator coordinator = createSavepointCoordinator(
				jobId,
				checkpointTimeout,
				vertices,
				vertices,
				vertices,
				checkpointIdCounter,
				savepointStore);
		// Trigger the savepoint
		Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
		assertFalse(savepointPathFuture.isCompleted());
		long checkpointId = checkpointIdCounter.getLastReturnedCount();
		assertEquals(0, checkpointId);
		// Verify send trigger messages
		for (ExecutionVertex vertex : vertices) {
			verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
		}
		PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
				.get(checkpointId);
		verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
				timestamp, 0, 2, 0, false, false);
		// Acknowledge tasks
		for (ExecutionVertex vertex : vertices) {
			coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
					jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
					checkpointId, createSerializedStateHandle(vertex), 0));
		}
		// The pending checkpoint is completed
		assertTrue(pendingCheckpoint.isDiscarded());
		assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
		// Verify send notify complete messages
		for (ExecutionVertex vertex : vertices) {
			verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
		}
		// Verify that the future has been completed
		assertTrue(savepointPathFuture.isCompleted());
		String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
		// Verify all promises removed
		assertEquals(0, getSavepointPromises(coordinator).size());
		coordinator.shutdown();
		// Verify the savepoint is still available
		CompletedCheckpoint savepoint = savepointStore.getState(savepointPath);
		verifySavepoint(savepoint, jobId, checkpointId, timestamp,
				vertices);
	}
	finally {
		FileUtils.deleteDirectory(tmpDir);
	}
}