@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)); }
@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()); } }
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; }
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; }
@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()); } }
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; }
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; }
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; }
@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()); } }
@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)); } }
@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)); }
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; }
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())); } }
@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)); }