class PendingCheckpoint {

	public TaskAcknowledgeResult acknowledgeCoordinatorState(
			OperatorInfo coordinatorInfo,
			@Nullable ByteStreamStateHandle stateHandle) {

		synchronized (lock) {
			if (discarded) {
				return TaskAcknowledgeResult.DISCARDED;
			}

			final OperatorID operatorId = coordinatorInfo.operatorId();
			OperatorState operatorState = operatorStates.get(operatorId);

			// sanity check for better error reporting
			if (!notYetAcknowledgedOperatorCoordinators.remove(operatorId)) {
				return operatorState != null && operatorState.getCoordinatorState() != null
						? TaskAcknowledgeResult.DUPLICATE
						: TaskAcknowledgeResult.UNKNOWN;
			}

			if (stateHandle != null) {
				if (operatorState == null) {
					operatorState = new OperatorState(
						operatorId, coordinatorInfo.currentParallelism(), coordinatorInfo.maxParallelism());
					operatorStates.put(operatorId, operatorState);
				}
				operatorState.setCoordinatorState(stateHandle);
			}

			return TaskAcknowledgeResult.SUCCESS;
		}
	}

	public  PendingCheckpoint(
			JobID jobId,
			long checkpointId,
			long checkpointTimestamp,
			Map<ExecutionAttemptID, ExecutionVertex> verticesToConfirm,
			Collection<OperatorID> operatorCoordinatorsToConfirm,
			Collection<String> masterStateIdentifiers,
			CheckpointProperties props,
			CheckpointStorageLocation targetLocation,
			Executor executor,
			CompletableFuture<CompletedCheckpoint> onCompletionPromise);

}