class SavepointCoordinator extends CheckpointCoordinator {

	public Future<String> triggerSavepoint(long timestamp) throws Exception {
		final Promise<String> promise = new scala.concurrent.impl.Promise.DefaultPromise<>();

		try {
			// Get the checkpoint ID up front. If we fail to trigger the checkpoint,
			// the ID will have changed, but this is OK as long as the checkpoint ID
			// generates ascending IDs.
			final long checkpointId = getAndIncrementCheckpointId();

			if (checkpointId == -1) {
				throw new IllegalStateException("Failed to get checkpoint Id");
			}

			// Important: make sure to add the promise to the map before calling
			// any methods that might trigger callbacks, which require the promise.
			// Otherwise, the might be race conditions.
			if (savepointPromises.put(checkpointId, promise) == null) {
				boolean success = false;

				try {
					// All good. The future will be completed as soon as the
					// triggered checkpoint is done.
					success = triggerCheckpoint(timestamp, checkpointId);
				}
				finally {
					if (!success) {
						savepointPromises.remove(checkpointId);
						promise.failure(new Exception("Failed to trigger savepoint"));
					}
				}
			}
			else {
				throw new IllegalStateException("Duplicate checkpoint ID");
			}
		}
		catch (Throwable t) {
			promise.failure(new Exception("Failed to trigger savepoint", t));
		}

		return promise.future();
	}

	public  SavepointCoordinator(
			JobID jobId,
			long baseInterval,
			long checkpointTimeout,
			ExecutionVertex[] tasksToTrigger,
			ExecutionVertex[] tasksToWaitFor,
			ExecutionVertex[] tasksToCommitTo,
			ClassLoader userClassLoader,
			CheckpointIDCounter checkpointIDCounter,
			StateStore<CompletedCheckpoint> savepointStore,
			CheckpointStatsTracker statsTracker);

}