Java 类org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptContainerAssignedEvent 实例源码

项目：hadoop 文件：MRApp.java

@Override
public void handle(ContainerAllocatorEvent event) {
  ContainerId cId =
      ContainerId.newContainerId(getContext().getApplicationAttemptId(),
        containerCount++);
  NodeId nodeId = NodeId.newInstance(NM_HOST, NM_PORT);
  Resource resource = Resource.newInstance(1234, 2, 2);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, nodeId.toString(), "user",
      resource, System.currentTimeMillis() + 10000, 42, 42,
      Priority.newInstance(0), 0);
  Token containerToken = newContainerToken(nodeId, "password".getBytes(),
        containerTokenIdentifier);
  Container container = Container.newInstance(cId, nodeId,
      NM_HOST + ":" + NM_HTTP_PORT, resource, null, containerToken);
  JobID id = TypeConverter.fromYarn(applicationId);
  JobId jobId = TypeConverter.toYarn(id);
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.REDUCE,
      100)));
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.MAP,
      100)));
  getContext().getEventHandler().handle(
      new TaskAttemptContainerAssignedEvent(event.getAttemptID(),
          container, null));
}

项目：hadoop 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hadoop 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      } else if (event instanceof JobEvent) {
        jobEvents.add((JobEvent)event);
      }
    }
  });
  return context;
}

项目：hadoop 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  try {
    super.heartbeat();
  } catch (Exception e) {
    LOG.error("error in heartbeat ", e);
    throw new YarnRuntimeException(e);
  }

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：aliyun-oss-hadoop-fs 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.debug("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：aliyun-oss-hadoop-fs 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      } else if (event instanceof JobEvent) {
        jobEvents.add((JobEvent)event);
      }
    }
  });
  return context;
}

项目：aliyun-oss-hadoop-fs 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  super.heartbeat();

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：big-c 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：big-c 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      } else if (event instanceof JobEvent) {
        jobEvents.add((JobEvent)event);
      }
    }
  });
  return context;
}

项目：big-c 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  try {
    super.heartbeat();
  } catch (Exception e) {
    LOG.error("error in heartbeat ", e);
    throw new YarnRuntimeException(e);
  }

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：hadoop-2.6.0-cdh5.4.3 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：hadoop-2.6.0-cdh5.4.3 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  try {
    super.heartbeat();
  } catch (Exception e) {
    LOG.error("error in heartbeat ", e);
    throw new YarnRuntimeException(e);
  }

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：hadoop-plus 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hadoop-plus 文件：TestRMContainerAllocator.java

private
    List<TaskAttemptContainerAssignedEvent> getContainerOnHost(JobId jobId,
        int taskAttemptId, int memory, String[] hosts, MockNM mockNM,
        DrainDispatcher dispatcher, MyContainerAllocator allocator)
        throws Exception {
  ContainerRequestEvent reqEvent =
      createReq(jobId, taskAttemptId, memory, hosts);
  allocator.sendRequest(reqEvent);

  // Send the request to the RM
  List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule();
  dispatcher.await();
  Assert.assertEquals("No of assignments must be 0", 0, assigned.size());

  // Heartbeat from the required nodeManager
  mockNM.nodeHeartbeat(true);
  dispatcher.await();

  assigned = allocator.schedule();
  dispatcher.await();
  return assigned;
}

项目：hadoop-plus 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：FlexMap 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  LOG.info("host is:"+allocated.getNodeId().getHost());

  //send the container to JobIml to provision data to this task
  eventHandler.handle(new JobTaskAttemptContainerAssinged(assigned.attemptID,allocated));
  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：FlexMap 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：FlexMap 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  try {
    super.heartbeat();
  } catch (Exception e) {
    LOG.error("error in heartbeat ", e);
    throw new YarnRuntimeException(e);
  }

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：hops 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.debug("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hops 文件：TestRMContainerAllocator.java

public List<TaskAttemptContainerAssignedEvent> schedule()
    throws Exception {
  // before doing heartbeat with RM, drain all the outstanding events to
  // ensure all the requests before this heartbeat is to be handled
  GenericTestUtils.waitFor(new Supplier<Boolean>() {
    public Boolean get() {
      return eventQueue.isEmpty();
    }
  }, 100, 10000);
  // run the scheduler
  super.heartbeat();

  List<TaskAttemptContainerAssignedEvent> result
    = new ArrayList<TaskAttemptContainerAssignedEvent>(events);
  events.clear();
  return result;
}

项目：hadoop-TCP 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hadoop-TCP 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：hardfs 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hardfs 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：hadoop-on-lustre2 文件：RMContainerAllocator.java

@SuppressWarnings("unchecked")
private void containerAssigned(Container allocated, 
                                ContainerRequest assigned) {
  // Update resource requests
  decContainerReq(assigned);

  // send the container-assigned event to task attempt
  eventHandler.handle(new TaskAttemptContainerAssignedEvent(
      assigned.attemptID, allocated, applicationACLs));

  assignedRequests.add(allocated, assigned.attemptID);

  if (LOG.isDebugEnabled()) {
    LOG.info("Assigned container (" + allocated + ") "
        + " to task " + assigned.attemptID + " on node "
        + allocated.getNodeId().toString());
  }
}

项目：hadoop-on-lustre2 文件：TestRMContainerAllocator.java

private static AppContext createAppContext(
    ApplicationAttemptId appAttemptId, Job job) {
  AppContext context = mock(AppContext.class);
  ApplicationId appId = appAttemptId.getApplicationId();
  when(context.getApplicationID()).thenReturn(appId);
  when(context.getApplicationAttemptId()).thenReturn(appAttemptId);
  when(context.getJob(isA(JobId.class))).thenReturn(job);
  when(context.getClusterInfo()).thenReturn(
    new ClusterInfo(Resource.newInstance(10240, 1)));
  when(context.getEventHandler()).thenReturn(new EventHandler() {
    @Override
    public void handle(Event event) {
      // Only capture interesting events.
      if (event instanceof TaskAttemptContainerAssignedEvent) {
        events.add((TaskAttemptContainerAssignedEvent) event);
      } else if (event instanceof TaskAttemptKillEvent) {
        taskAttemptKillEvents.add((TaskAttemptKillEvent)event);
      } else if (event instanceof JobUpdatedNodesEvent) {
        jobUpdatedNodeEvents.add((JobUpdatedNodesEvent)event);
      }
    }
  });
  return context;
}

项目：hadoop 文件：LocalContainerAllocator.java

@SuppressWarnings("unchecked")
@Override
public void handle(ContainerAllocatorEvent event) {
  if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
    LOG.info("Processing the event " + event.toString());
    // Assign the same container ID as the AM
    ContainerId cID =
        ContainerId.newContainerId(getContext().getApplicationAttemptId(),
          this.containerId.getContainerId());
    Container container = recordFactory.newRecordInstance(Container.class);
    container.setId(cID);
    NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
    container.setNodeId(nodeId);
    container.setContainerToken(null);
    container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
    // send the container-assigned event to task attempt

    if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
      JobCounterUpdateEvent jce =
          new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
              .getJobId());
      // TODO Setting OTHER_LOCAL_MAP for now.
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
    }
    eventHandler.handle(new TaskAttemptContainerAssignedEvent(
        event.getAttemptID(), container, applicationACLs));
  }
}

项目：hadoop 文件：TaskAttemptImpl.java

@SuppressWarnings({ "unchecked" })
@Override
public void transition(final TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  final TaskAttemptContainerAssignedEvent cEvent = 
    (TaskAttemptContainerAssignedEvent) event;
  Container container = cEvent.getContainer();
  taskAttempt.container = container;
  // this is a _real_ Task (classic Hadoop mapred flavor):
  taskAttempt.remoteTask = taskAttempt.createRemoteTask();
  taskAttempt.jvmID =
      new WrappedJvmID(taskAttempt.remoteTask.getTaskID().getJobID(),
          taskAttempt.remoteTask.isMapTask(),
          taskAttempt.container.getId().getContainerId());
  taskAttempt.taskAttemptListener.registerPendingTask(
      taskAttempt.remoteTask, taskAttempt.jvmID);

  taskAttempt.computeRackAndLocality();

  //launch the container
  //create the container object to be launched for a given Task attempt
  ContainerLaunchContext launchContext = createContainerLaunchContext(
      cEvent.getApplicationACLs(), taskAttempt.conf, taskAttempt.jobToken,
      taskAttempt.remoteTask, taskAttempt.oldJobId, taskAttempt.jvmID,
      taskAttempt.taskAttemptListener, taskAttempt.credentials);
  taskAttempt.eventHandler
    .handle(new ContainerRemoteLaunchEvent(taskAttempt.attemptId,
      launchContext, container, taskAttempt.remoteTask));

  // send event to speculator that our container needs are satisfied
  taskAttempt.eventHandler.handle
      (new SpeculatorEvent(taskAttempt.getID().getTaskId(), -1));
}

项目：hadoop 文件：TestRMContainerAllocator.java

private
    List<TaskAttemptContainerAssignedEvent> getContainerOnHost(JobId jobId,
        int taskAttemptId, int memory, String[] hosts, MockNM mockNM,
        DrainDispatcher dispatcher, MyContainerAllocator allocator,
        int expectedAdditions1, int expectedRemovals1,
        int expectedAdditions2, int expectedRemovals2, MyResourceManager rm)
        throws Exception {
  ContainerRequestEvent reqEvent =
      createReq(jobId, taskAttemptId, memory, hosts);
  allocator.sendRequest(reqEvent);

  // Send the request to the RM
  List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions1, expectedRemovals1, rm);
  Assert.assertEquals("No of assignments must be 0", 0, assigned.size());

  // Heartbeat from the required nodeManager
  mockNM.nodeHeartbeat(true);
  dispatcher.await();

  assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions2, expectedRemovals2, rm);
  return assigned;
}

项目：hadoop 文件：TestRMContainerAllocator.java

private void checkAssignment(ContainerRequestEvent request,
    TaskAttemptContainerAssignedEvent assigned, boolean checkHostMatch) {
  Assert.assertNotNull("Nothing assigned to attempt "
      + request.getAttemptID(), assigned);
  Assert.assertEquals("assigned to wrong attempt", request.getAttemptID(),
      assigned.getTaskAttemptID());
  if (checkHostMatch) {
    Assert.assertTrue("Not assigned to requested host", Arrays.asList(
        request.getHosts()).contains(
        assigned.getContainer().getNodeId().getHost()));
  }
}

项目：aliyun-oss-hadoop-fs 文件：LocalContainerAllocator.java

@SuppressWarnings("unchecked")
@Override
public void handle(ContainerAllocatorEvent event) {
  if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
    LOG.info("Processing the event " + event.toString());
    // Assign the same container ID as the AM
    ContainerId cID =
        ContainerId.newContainerId(getContext().getApplicationAttemptId(),
          this.containerId.getContainerId());
    Container container = recordFactory.newRecordInstance(Container.class);
    container.setId(cID);
    NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
    container.setNodeId(nodeId);
    container.setContainerToken(null);
    container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
    // send the container-assigned event to task attempt

    if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
      JobCounterUpdateEvent jce =
          new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
              .getJobId());
      // TODO Setting OTHER_LOCAL_MAP for now.
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
    }
    eventHandler.handle(new TaskAttemptContainerAssignedEvent(
        event.getAttemptID(), container, applicationACLs));
  }
}

项目：aliyun-oss-hadoop-fs 文件：TaskAttemptImpl.java

@SuppressWarnings({ "unchecked" })
@Override
public void transition(final TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  final TaskAttemptContainerAssignedEvent cEvent = 
    (TaskAttemptContainerAssignedEvent) event;
  Container container = cEvent.getContainer();
  taskAttempt.container = container;
  // this is a _real_ Task (classic Hadoop mapred flavor):
  taskAttempt.remoteTask = taskAttempt.createRemoteTask();
  taskAttempt.jvmID =
      new WrappedJvmID(taskAttempt.remoteTask.getTaskID().getJobID(),
          taskAttempt.remoteTask.isMapTask(),
          taskAttempt.container.getId().getContainerId());
  taskAttempt.taskAttemptListener.registerPendingTask(
      taskAttempt.remoteTask, taskAttempt.jvmID);

  taskAttempt.computeRackAndLocality();

  //launch the container
  //create the container object to be launched for a given Task attempt
  ContainerLaunchContext launchContext = createContainerLaunchContext(
      cEvent.getApplicationACLs(), taskAttempt.conf, taskAttempt.jobToken,
      taskAttempt.remoteTask, taskAttempt.oldJobId, taskAttempt.jvmID,
      taskAttempt.taskAttemptListener, taskAttempt.credentials);
  taskAttempt.eventHandler
    .handle(new ContainerRemoteLaunchEvent(taskAttempt.attemptId,
      launchContext, container, taskAttempt.remoteTask));

  // send event to speculator that our container needs are satisfied
  taskAttempt.eventHandler.handle
      (new SpeculatorEvent(taskAttempt.getID().getTaskId(), -1));
}

项目：aliyun-oss-hadoop-fs 文件：MRApp.java

@Override
public void handle(ContainerAllocatorEvent event) {
  ContainerId cId =
      ContainerId.newContainerId(getContext().getApplicationAttemptId(),
        containerCount++);
  NodeId nodeId = NodeId.newInstance(NM_HOST, NM_PORT);
  Resource resource = Resource.newInstance(1234, 2);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, nodeId.toString(), "user",
      resource, System.currentTimeMillis() + 10000, 42, 42,
      Priority.newInstance(0), 0);
  Token containerToken = newContainerToken(nodeId, "password".getBytes(),
        containerTokenIdentifier);
  Container container = Container.newInstance(cId, nodeId,
      NM_HOST + ":" + NM_HTTP_PORT, resource, null, containerToken);
  JobID id = TypeConverter.fromYarn(applicationId);
  JobId jobId = TypeConverter.toYarn(id);
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.REDUCE,
      100)));
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.MAP,
      100)));
  getContext().getEventHandler().handle(
      new TaskAttemptContainerAssignedEvent(event.getAttemptID(),
          container, null));
}

项目：aliyun-oss-hadoop-fs 文件：TestRMContainerAllocator.java

private
    List<TaskAttemptContainerAssignedEvent> getContainerOnHost(JobId jobId,
        int taskAttemptId, int memory, String[] hosts, MockNM mockNM,
        DrainDispatcher dispatcher, MyContainerAllocator allocator,
        int expectedAdditions1, int expectedRemovals1,
        int expectedAdditions2, int expectedRemovals2, MyResourceManager rm)
        throws Exception {
  ContainerRequestEvent reqEvent =
      createReq(jobId, taskAttemptId, memory, hosts);
  allocator.sendRequest(reqEvent);

  // Send the request to the RM
  List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions1, expectedRemovals1, rm);
  Assert.assertEquals("No of assignments must be 0", 0, assigned.size());

  // Heartbeat from the required nodeManager
  mockNM.nodeHeartbeat(true);
  dispatcher.await();

  assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions2, expectedRemovals2, rm);
  return assigned;
}

项目：aliyun-oss-hadoop-fs 文件：TestRMContainerAllocator.java

private void checkAssignment(ContainerRequestEvent request,
    TaskAttemptContainerAssignedEvent assigned, boolean checkHostMatch) {
  Assert.assertNotNull("Nothing assigned to attempt "
      + request.getAttemptID(), assigned);
  Assert.assertEquals("assigned to wrong attempt", request.getAttemptID(),
      assigned.getTaskAttemptID());
  if (checkHostMatch) {
    Assert.assertTrue("Not assigned to requested host", Arrays.asList(
        request.getHosts()).contains(
        assigned.getContainer().getNodeId().getHost()));
  }
}

项目：big-c 文件：LocalContainerAllocator.java

@SuppressWarnings("unchecked")
@Override
public void handle(ContainerAllocatorEvent event) {
  if (event.getType() == ContainerAllocator.EventType.CONTAINER_REQ) {
    LOG.info("Processing the event " + event.toString());
    // Assign the same container ID as the AM
    ContainerId cID =
        ContainerId.newContainerId(getContext().getApplicationAttemptId(),
          this.containerId.getContainerId());
    Container container = recordFactory.newRecordInstance(Container.class);
    container.setId(cID);
    NodeId nodeId = NodeId.newInstance(this.nmHost, this.nmPort);
    container.setNodeId(nodeId);
    container.setContainerToken(null);
    container.setNodeHttpAddress(this.nmHost + ":" + this.nmHttpPort);
    // send the container-assigned event to task attempt

    if (event.getAttemptID().getTaskId().getTaskType() == TaskType.MAP) {
      JobCounterUpdateEvent jce =
          new JobCounterUpdateEvent(event.getAttemptID().getTaskId()
              .getJobId());
      // TODO Setting OTHER_LOCAL_MAP for now.
      jce.addCounterUpdate(JobCounter.OTHER_LOCAL_MAPS, 1);
      eventHandler.handle(jce);
    }
    eventHandler.handle(new TaskAttemptContainerAssignedEvent(
        event.getAttemptID(), container, applicationACLs));
  }
}

项目：big-c 文件：TaskAttemptImpl.java

@SuppressWarnings({ "unchecked" })
@Override
public void transition(final TaskAttemptImpl taskAttempt, 
    TaskAttemptEvent event) {
  final TaskAttemptContainerAssignedEvent cEvent = 
    (TaskAttemptContainerAssignedEvent) event;
  Container container = cEvent.getContainer();
  taskAttempt.container = container;
  // this is a _real_ Task (classic Hadoop mapred flavor):
  taskAttempt.remoteTask = taskAttempt.createRemoteTask();
  taskAttempt.jvmID =
      new WrappedJvmID(taskAttempt.remoteTask.getTaskID().getJobID(),
          taskAttempt.remoteTask.isMapTask(),
          taskAttempt.container.getId().getContainerId());
  taskAttempt.taskAttemptListener.registerPendingTask(
      taskAttempt.remoteTask, taskAttempt.jvmID);

  taskAttempt.computeRackAndLocality();

  //launch the container
  //create the container object to be launched for a given Task attempt
  ContainerLaunchContext launchContext = createContainerLaunchContext(
      cEvent.getApplicationACLs(), taskAttempt.conf, taskAttempt.jobToken,
      taskAttempt.remoteTask, taskAttempt.oldJobId, taskAttempt.jvmID,
      taskAttempt.taskAttemptListener, taskAttempt.credentials);
  taskAttempt.eventHandler
    .handle(new ContainerRemoteLaunchEvent(taskAttempt.attemptId,
      launchContext, container, taskAttempt.remoteTask));

  // send event to speculator that our container needs are satisfied
  taskAttempt.eventHandler.handle
      (new SpeculatorEvent(taskAttempt.getID().getTaskId(), -1));
}

项目：big-c 文件：MRApp.java

@Override
public void handle(ContainerAllocatorEvent event) {
  ContainerId cId =
      ContainerId.newContainerId(getContext().getApplicationAttemptId(),
        containerCount++);
  NodeId nodeId = NodeId.newInstance(NM_HOST, NM_PORT);
  Resource resource = Resource.newInstance(1234, 2);
  ContainerTokenIdentifier containerTokenIdentifier =
      new ContainerTokenIdentifier(cId, nodeId.toString(), "user",
      resource, System.currentTimeMillis() + 10000, 42, 42,
      Priority.newInstance(0), 0);
  Token containerToken = newContainerToken(nodeId, "password".getBytes(),
        containerTokenIdentifier);
  Container container = Container.newInstance(cId, nodeId,
      NM_HOST + ":" + NM_HTTP_PORT, resource, null, containerToken);
  JobID id = TypeConverter.fromYarn(applicationId);
  JobId jobId = TypeConverter.toYarn(id);
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.REDUCE,
      100)));
  getContext().getEventHandler().handle(new JobHistoryEvent(jobId, 
      new NormalizedResourceEvent(
          org.apache.hadoop.mapreduce.TaskType.MAP,
      100)));
  getContext().getEventHandler().handle(
      new TaskAttemptContainerAssignedEvent(event.getAttemptID(),
          container, null));
}

项目：big-c 文件：TestRMContainerAllocator.java

private
    List<TaskAttemptContainerAssignedEvent> getContainerOnHost(JobId jobId,
        int taskAttemptId, int memory, String[] hosts, MockNM mockNM,
        DrainDispatcher dispatcher, MyContainerAllocator allocator,
        int expectedAdditions1, int expectedRemovals1,
        int expectedAdditions2, int expectedRemovals2, MyResourceManager rm)
        throws Exception {
  ContainerRequestEvent reqEvent =
      createReq(jobId, taskAttemptId, memory, hosts);
  allocator.sendRequest(reqEvent);

  // Send the request to the RM
  List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions1, expectedRemovals1, rm);
  Assert.assertEquals("No of assignments must be 0", 0, assigned.size());

  // Heartbeat from the required nodeManager
  mockNM.nodeHeartbeat(true);
  dispatcher.await();

  assigned = allocator.schedule();
  dispatcher.await();
  assertBlacklistAdditionsAndRemovals(
      expectedAdditions2, expectedRemovals2, rm);
  return assigned;
}