@Test
public void testAbortSavepointsOnShutdown() throws Exception {

	JobID jobId = new JobID();
	ExecutionVertex[] vertices = new ExecutionVertex[] {
			mockExecutionVertex(jobId),
			mockExecutionVertex(jobId) };
	SavepointCoordinator coordinator = createSavepointCoordinator(
			jobId,
			60 * 1000,
			vertices,
			vertices,
			vertices,
			new MockCheckpointIdCounter(),
			new HeapStateStore<CompletedCheckpoint>());
	// Trigger savepoints
	List<Future<String>> savepointPathFutures = new ArrayList<>();
	savepointPathFutures.add(coordinator.triggerSavepoint(12731273));
	savepointPathFutures.add(coordinator.triggerSavepoint(12731273 + 123));
	for (Future<String> future : savepointPathFutures) {
		assertFalse(future.isCompleted());
	}
	coordinator.shutdown();
	// Verify futures failed
	for (Future<String> future : savepointPathFutures) {
		assertTrue(future.isCompleted());
		try {
			Await.result(future, FiniteDuration.Zero());
			fail("Did not throw expected Exception after shutdown");
		}
		catch (Exception ignored) {
		}
	}
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
}