class PendingSavepoint extends PendingCheckpoint {

	@Override
	public CompletedCheckpoint finalizeCheckpoint() throws Exception {
		// finalize checkpoint (this also disposes this pending checkpoint)
		CompletedCheckpoint completedCheckpoint = super.finalizeCheckpoint();

		// now store the checkpoint externally as a savepoint
		try {
			Savepoint savepoint = new SavepointV1(
					completedCheckpoint.getCheckpointID(),
					completedCheckpoint.getTaskStates().values());
			
			String path = store.storeSavepoint(savepoint);
			onCompletionPromise.success(path);
		}
		catch (Throwable t) {
			LOG.warn("Failed to store savepoint.", t);
			onCompletionPromise.failure(t);

			ExceptionUtils.rethrow(t, "Failed to store savepoint.");
		}

		return completedCheckpoint;
	}

	public  PendingSavepoint(
			JobID jobId,
			long checkpointId,
			long checkpointTimestamp,
			Map<ExecutionAttemptID, ExecutionVertex> verticesToConfirm,
			SavepointStore store);

}