code: public TaskMetricGroup addTask(
  			final JobVertexID jobVertexId,
  			final ExecutionAttemptID executionAttemptID,
  			final String taskName,
  			final int subtaskIndex,
  			final int attemptNumber) {
  		checkNotNull(jobVertexId);
  		checkNotNull(executionAttemptID);
  		checkNotNull(taskName);
  
  		synchronized (this) {
  			if (!isClosed()) {
  				TaskMetricGroup prior = tasks.get(executionAttemptID);
  				if (prior != null) {
  					return prior;
  				} else {
  					TaskMetricGroup task = new TaskMetricGroup(
  						registry,
  						this,
  						jobVertexId,
  						executionAttemptID,
  						taskName,
  						subtaskIndex,
  						attemptNumber);
  					tasks.put(executionAttemptID, task);
  					return task;
  				}
  			} else {
  				return null;
  			}
  		}
  	}
docs: ------------------------------------------------------------------------