code: private ConcurrentHashMap<JobVertexID, TaskStateStats> createEmptyTaskStateStatsMap() {
  		ConcurrentHashMap<JobVertexID, TaskStateStats> taskStatsMap = new ConcurrentHashMap<>(jobVertices.size());
  		for (ExecutionJobVertex vertex : jobVertices) {
  			TaskStateStats taskStats = new TaskStateStats(vertex.getJobVertexId(), vertex.getParallelism());
  			taskStatsMap.put(vertex.getJobVertexId(), taskStats);
  		}
  		return taskStatsMap;
  	}
docs: Creates an empty map with a {@link TaskStateStats} instance per task
  that is involved in the checkpoint.
  
  @return An empty map with an {@link TaskStateStats} entry for each task that is involved in the checkpoint.