@Test
public void testAbortSavepointIfAckTasksAreNotExecuted() throws Exception {

	JobID jobId = new JobID();
	ExecutionVertex[] triggerVertices = new ExecutionVertex[] {
			mockExecutionVertex(jobId),
			mockExecutionVertex(jobId) };
	ExecutionVertex[] ackVertices = new ExecutionVertex[] {
			mock(ExecutionVertex.class),
			mock(ExecutionVertex.class) };
	SavepointCoordinator coordinator = createSavepointCoordinator(
			jobId,
			60 * 1000,
			triggerVertices,
			ackVertices,
			new ExecutionVertex[] {},
			new MockCheckpointIdCounter(),
			new HeapStateStore<CompletedCheckpoint>());
	// Trigger savepoint
	Future<String> savepointPathFuture = coordinator.triggerSavepoint(1238123);
	// Abort the savepoint, because the vertices are not running
	assertTrue(savepointPathFuture.isCompleted());
	try {
		Await.result(savepointPathFuture, FiniteDuration.Zero());
		fail("Did not throw expected Exception after shutdown");
	}
	catch (Exception ignored) {
	}
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
	coordinator.shutdown();
}