@Test
public void testSimpleTriggerSavepoint() throws Exception {

	JobID jobId = new JobID();
	long checkpointTimeout = 60 * 1000;
	long timestamp = 1272635;
	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 savepoint
	Future<String> savepointPathFuture = coordinator.triggerSavepoint(timestamp);
	assertFalse(savepointPathFuture.isCompleted());
	long checkpointId = checkpointIdCounter.getLastReturnedCount();
	assertEquals(0, checkpointId);
	// Verify send trigger messages
	for (ExecutionVertex vertex : vertices) {
		verifyTriggerCheckpoint(vertex, checkpointId, timestamp);
	}
	PendingCheckpoint pendingCheckpoint = coordinator.getPendingCheckpoints()
			.get(checkpointId);
	verifyPendingCheckpoint(pendingCheckpoint, jobId, checkpointId,
			timestamp, 0, 2, 0, false, false);
	// Acknowledge tasks
	for (ExecutionVertex vertex : vertices) {
		coordinator.receiveAcknowledgeMessage(new AcknowledgeCheckpoint(
				jobId, vertex.getCurrentExecutionAttempt().getAttemptId(),
				checkpointId, createSerializedStateHandle(vertex), 0));
	}
	// The pending checkpoint is completed
	assertTrue(pendingCheckpoint.isDiscarded());
	assertEquals(0, coordinator.getSuccessfulCheckpoints().size());
	// Verify send notify complete messages
	for (ExecutionVertex vertex : vertices) {
		verifyNotifyCheckpointComplete(vertex, checkpointId, timestamp);
	}
	// Verify that the future has been completed
	assertTrue(savepointPathFuture.isCompleted());
	String savepointPath = Await.result(savepointPathFuture, FiniteDuration.Zero());
	// Verify the savepoint
	CompletedCheckpoint savepoint = savepointStore.getState(savepointPath);
	verifySavepoint(savepoint, jobId, checkpointId, timestamp,
			vertices);
	// Verify all promises removed
	assertEquals(0, getSavepointPromises(coordinator).size());
	coordinator.shutdown();
}