docs: @Override
  	public void acknowledgeCheckpoint(
  			final JobID jobID,
  			final ExecutionAttemptID executionAttemptID,
  			final long checkpointId,
  			final CheckpointMetrics checkpointMetrics,
  			final TaskStateSnapshot checkpointState) {
  
  		final CheckpointCoordinator checkpointCoordinator = executionGraph.getCheckpointCoordinator();
  		final AcknowledgeCheckpoint ackMessage = new AcknowledgeCheckpoint(
  			jobID,
  			executionAttemptID,
  			checkpointId,
  			checkpointMetrics,
  			checkpointState);
  
  		if (checkpointCoordinator != null) {
  			getRpcService().execute(() -> {
  				try {
  					checkpointCoordinator.receiveAcknowledgeMessage(ackMessage);
  				} catch (Throwable t) {
  					log.warn("Error while processing checkpoint acknowledgement message", t);
  				}
  			});
  		} else {
  			String errorMessage = "Received AcknowledgeCheckpoint message for job {} with no CheckpointCoordinator";
  			if (executionGraph.getState() == JobStatus.RUNNING) {
  				log.error(errorMessage, jobGraph.getJobID());
  			} else {
  				log.debug(errorMessage, jobGraph.getJobID());
  			}
  		}
  	}