code: private Supervisor createSupervisor(final Job job, final Map<String, Integer> portAllocation) {
      log.debug("creating job supervisor: {}", job);
      final TaskStatus taskStatus = model.getTaskStatus(job.getId());
      final String containerId = (taskStatus == null) ? null : taskStatus.getContainerId();
      final Supervisor supervisor = supervisorFactory.create(job, containerId, portAllocation,
          supervisorListener);
      supervisors.put(job.getId(), supervisor);
      return supervisor;
    }
docs: Create a job supervisor.
  
  @param job The job .