@Test
public void testAbortOnCheckpointTimeout() throws Exception {

	JobID jobId = new JobID();
	ExecutionVertex[] vertices = new ExecutionVertex[] {
			mockExecutionVertex(jobId),
			mockExecutionVertex(jobId) };
	ExecutionVertex commitVertex = mockExecutionVertex(jobId);
	MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
	SavepointCoordinator coordinator = createSavepointCoordinator(
			jobId,
			20,
			vertices,
			vertices,
			new ExecutionVertex[] { commitVertex },
			checkpointIdCounter,
			new HeapStateStore<CompletedCheckpoint>());
	// Trigger the savepoint
	Future<String> savepointPathFuture = coordinator.triggerSavepoint(12731273);
	assertFalse(savepointPathFuture.isCompleted());
	long checkpointId = checkpointIdCounter.getLastReturnedCount();
	// Acknowledge single task
	coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
			jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
			checkpointId, createSerializedStateHandle(vertices[0]), 0));
	PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
			.get(checkpointId);
	assertFalse(pendingCheckpoint.isDiscarded());
	// Wait for savepoint to timeout
	Deadline deadline = FiniteDuration.apply(5, "s").fromNow();
	while (deadline.hasTimeLeft()
			&& !pendingCheckpoint.isDiscarded()
			&& coordinator.getNumberOfPendingCheckpoints() > 0) {
		Thread.sleep(250);
	}
	// Verify discarded
	assertTrue(pendingCheckpoint.isDiscarded());
	assertEquals(0, coordinator.getNumberOfPendingCheckpoints());
	assertEquals(0, coordinator.getNumberOfRetainedSuccessfulCheckpoints());
	// No commit for timeout
	verify(commitVertex, times(0)).sendMessageToCurrentExecution(
			any(NotifyCheckpointComplete.class), any(ExecutionAttemptID.class));
	assertTrue(savepointPathFuture.isCompleted());
	try {
		Await.result(savepointPathFuture, FiniteDuration.Zero());
		fail("Did not throw expected Exception after timeout");
	}
	catch (Exception ignored) {
	}
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
	coordinator.shutdown();
}