@Test
public void testAbortSavepointIfSubsumed() throws Exception {

	JobID jobId = new JobID();
	long checkpointTimeout = 60 * 1000;
	long[] timestamps = new long[] { 1272635, 1272635 + 10 };
	long[] checkpointIds = new long[2];
	ExecutionVertex[] vertices = new ExecutionVertex[] {
			mockExecutionVertex(jobId),
			mockExecutionVertex(jobId) };
	MockCheckpointIdCounter checkpointIdCounter = new MockCheckpointIdCounter();
	HeapStateStore<CompletedCheckpoint> savepointStore = new HeapStateStore<>();
	SavepointCoordinator coordinator = createSavepointCoordinator(
			jobId,
			checkpointTimeout,
			vertices,
			vertices,
			vertices,
			checkpointIdCounter,
			savepointStore);
	// Trigger the savepoints
	List<Future<String>> savepointPathFutures = new ArrayList<>();
	savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[0]));
	checkpointIds[0] = checkpointIdCounter.getLastReturnedCount();
	savepointPathFutures.add(coordinator.triggerSavepoint(timestamps[1]));
	checkpointIds[1] = checkpointIdCounter.getLastReturnedCount();
	for (Future<String> future : savepointPathFutures) {
		assertFalse(future.isCompleted());
	}
	// Verify send trigger messages
	for (ExecutionVertex vertex : vertices) {
		verifyTriggerCheckpoint(vertex, checkpointIds[0], timestamps[0]);
		verifyTriggerCheckpoint(vertex, checkpointIds[1], timestamps[1]);
	}
	PendingCheckpoint[] pendingCheckpoints = new PendingCheckpoint[] {
			coordinator.getPendingCheckpoints().get(checkpointIds[0]),
			coordinator.getPendingCheckpoints().get(checkpointIds[1]) };
	verifyPendingCheckpoint(pendingCheckpoints[0], jobId, checkpointIds[0],
			timestamps[0], 0, 2, 0, false, false);
	verifyPendingCheckpoint(pendingCheckpoints[1], jobId, checkpointIds[1],
			timestamps[1], 0, 2, 0, false, false);
	// Acknowledge second checkpoint...
	for (ExecutionVertex vertex : vertices) {
		coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
				jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
				checkpointIds[1], createSerializedStateHandle(vertex), 0));
	}
	// ...and one task of first checkpoint
	coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
			jobId, vertices[0].getCurrentExecutionAttempt().getAttemptId(),
			checkpointIds[0], createSerializedStateHandle(vertices[0]), 0));
	// The second pending checkpoint is completed and subsumes the first one
	assertTrue(pendingCheckpoints[0].isDiscarded());
	assertTrue(pendingCheckpoints[1].isDiscarded());
	assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
	// Verify send notify complete messages for second checkpoint
	for (ExecutionVertex vertex : vertices) {
		verifyNotifyCheckpointComplete(vertex, checkpointIds[1], timestamps[1]);
	}
	CompletedCheckpoint[] savepoints = new CompletedCheckpoint[2];
	String[] savepointPaths = new String[2];
	// Verify that the futures have both been completed
	assertTrue(savepointPathFutures.get(0).isCompleted());
	try {
		savepointPaths[0] = Await.result(savepointPathFutures.get(0), FiniteDuration.Zero());
		fail("Did not throw expected exception");
	}
	catch (Exception ignored) {
	}
	// Verify the second savepoint
	assertTrue(savepointPathFutures.get(1).isCompleted());
	savepointPaths[1] = Await.result(savepointPathFutures.get(1), FiniteDuration.Zero());
	savepoints[1] = savepointStore.getState(savepointPaths[1]);
	verifySavepoint(savepoints[1], jobId, checkpointIds[1], timestamps[1],
			vertices);
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
	coordinator.shutdown();
}