docs: public static CheckpointStatistics generateCheckpointStatistics(AbstractCheckpointStats checkpointStats, boolean includeTaskCheckpointStatistics) {
  		Preconditions.checkNotNull(checkpointStats);
  
  		Map<JobVertexID, TaskCheckpointStatistics> checkpointStatisticsPerTask;
  
  		if (includeTaskCheckpointStatistics) {
  			Collection<TaskStateStats> taskStateStats = checkpointStats.getAllTaskStateStats();
  
  			checkpointStatisticsPerTask = new HashMap<>(taskStateStats.size());
  
  			for (TaskStateStats taskStateStat : taskStateStats) {
  				checkpointStatisticsPerTask.put(
  					taskStateStat.getJobVertexId(),
  					new TaskCheckpointStatistics(
  							checkpointStats.getCheckpointId(),
  							checkpointStats.getStatus(),
  						taskStateStat.getLatestAckTimestamp(),
  						taskStateStat.getStateSize(),
  						taskStateStat.getEndToEndDuration(checkpointStats.getTriggerTimestamp()),
  						taskStateStat.getAlignmentBuffered(),
  						taskStateStat.getNumberOfSubtasks(),
  						taskStateStat.getNumberOfAcknowledgedSubtasks()));
  			}
  		} else {
  			checkpointStatisticsPerTask = Collections.emptyMap();
  		}
  
  		if (checkpointStats instanceof CompletedCheckpointStats) {
  			final CompletedCheckpointStats completedCheckpointStats = ((CompletedCheckpointStats) checkpointStats);
  
  			return new CheckpointStatistics.CompletedCheckpointStatistics(
  				completedCheckpointStats.getCheckpointId(),
  				completedCheckpointStats.getStatus(),
  				completedCheckpointStats.getProperties().isSavepoint(),
  				completedCheckpointStats.getTriggerTimestamp(),
  				completedCheckpointStats.getLatestAckTimestamp(),
  				completedCheckpointStats.getStateSize(),
  				completedCheckpointStats.getEndToEndDuration(),
  				completedCheckpointStats.getAlignmentBuffered(),
  				completedCheckpointStats.getNumberOfSubtasks(),
  				completedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
  				checkpointStatisticsPerTask,
  				completedCheckpointStats.getExternalPath(),
  				completedCheckpointStats.isDiscarded());
  		} else if (checkpointStats instanceof FailedCheckpointStats) {
  			final FailedCheckpointStats failedCheckpointStats = ((FailedCheckpointStats) checkpointStats);
  
  			return new CheckpointStatistics.FailedCheckpointStatistics(
  				failedCheckpointStats.getCheckpointId(),
  				failedCheckpointStats.getStatus(),
  				failedCheckpointStats.getProperties().isSavepoint(),
  				failedCheckpointStats.getTriggerTimestamp(),
  				failedCheckpointStats.getLatestAckTimestamp(),
  				failedCheckpointStats.getStateSize(),
  				failedCheckpointStats.getEndToEndDuration(),
  				failedCheckpointStats.getAlignmentBuffered(),
  				failedCheckpointStats.getNumberOfSubtasks(),
  				failedCheckpointStats.getNumberOfAcknowledgedSubtasks(),
  				checkpointStatisticsPerTask,
  				failedCheckpointStats.getFailureTimestamp(),
  				failedCheckpointStats.getFailureMessage());
  		} else if (checkpointStats instanceof PendingCheckpointStats) {
  			final PendingCheckpointStats pendingCheckpointStats = ((PendingCheckpointStats) checkpointStats);
  
  			return new CheckpointStatistics.PendingCheckpointStatistics(
  				pendingCheckpointStats.getCheckpointId(),
  				pendingCheckpointStats.getStatus(),
  				pendingCheckpointStats.getProperties().isSavepoint(),
  				pendingCheckpointStats.getTriggerTimestamp(),
  				pendingCheckpointStats.getLatestAckTimestamp(),
  				pendingCheckpointStats.getStateSize(),
  				pendingCheckpointStats.getEndToEndDuration(),
  				pendingCheckpointStats.getAlignmentBuffered(),
  				pendingCheckpointStats.getNumberOfSubtasks(),
  				pendingCheckpointStats.getNumberOfAcknowledgedSubtasks(),
  				checkpointStatisticsPerTask
  			);
  		} else {
  			throw new IllegalArgumentException("Given checkpoint stats object of type "
  				+ checkpointStats.getClass().getName() + " cannot be converted.");
  		}
  	}