@Test
public void testAbortSavepointOnStateStoreFailure() throws Exception {

	JobID jobId = new JobID();
	ExecutionJobVertex jobVertex = mockExecutionJobVertex(jobId, new JobVertexID(), 4);
	HeapStateStore<CompletedCheckpoint> savepointStore = spy(
			new HeapStateStore<CompletedCheckpoint>());
	SavepointCoordinator coordinator = createSavepointCoordinator(
			jobId,
			60 * 1000,
			jobVertex.getTaskVertices(),
			jobVertex.getTaskVertices(),
			new ExecutionVertex[] {},
			new MockCheckpointIdCounter(),
			savepointStore);
	// Failure on putState
	doThrow(new Exception("TestException"))
			.when(savepointStore).putState(any(CompletedCheckpoint.class));
	Future<String> savepointPathFuture = coordinator.triggerSavepoint(1231273123);
	// Acknowledge all tasks
	for (ExecutionVertex vertex : jobVertex.getTaskVertices()) {
		ExecutionAttemptID attemptId = vertex.getCurrentExecutionAttempt().getAttemptId();
		coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
				jobId, attemptId, 0, createSerializedStateHandle(vertex), 0));
	}
	try {
		Await.result(savepointPathFuture, FiniteDuration.Zero());
		fail("Did not throw expected Exception after rollback with savepoint store failure.");
	}
	catch (Exception ignored) {
	}
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
	coordinator.shutdown();
}