code: public TaskMetricGroup addTaskForJob(
  			final JobID jobId,
  			final String jobName,
  			final JobVertexID jobVertexId,
  			final ExecutionAttemptID executionAttemptId,
  			final String taskName,
  			final int subtaskIndex,
  			final int attemptNumber) {
  		Preconditions.checkNotNull(jobId);
  
  		String resolvedJobName = jobName == null || jobName.isEmpty()
  			? jobId.toString()
  			: jobName;
  
  		// we cannot strictly lock both our map modification and the job group modification
  		// because it might lead to a deadlock
  		while (true) {
  			// get or create a jobs metric group
  			TaskManagerJobMetricGroup currentJobGroup;
  			synchronized (this) {
  				currentJobGroup = jobs.get(jobId);
  
  				if (currentJobGroup == null || currentJobGroup.isClosed()) {
  					currentJobGroup = new TaskManagerJobMetricGroup(registry, this, jobId, resolvedJobName);
  					jobs.put(jobId, currentJobGroup);
  				}
  			}
  
  			// try to add another task. this may fail if we found a pre-existing job metrics
  			// group and it is closed concurrently
  			TaskMetricGroup taskGroup = currentJobGroup.addTask(
  				jobVertexId,
  				executionAttemptId,
  				taskName,
  				subtaskIndex,
  				attemptNumber);
  
  			if (taskGroup != null) {
  				// successfully added the next task
  				return taskGroup;
  			}
  
  			// else fall through the loop
  		}
  	}
docs: ------------------------------------------------------------------------