@Test public void testMapNodeLocality() throws Exception { LOG.info("Running testMapNodeLocality"); Configuration conf = new Configuration(); MyResourceManager rm = new MyResourceManager(conf); rm.start(); DrainDispatcher dispatcher = (DrainDispatcher) rm.getRMContext() .getDispatcher(); RMApp app = rm.submitApp(1024); dispatcher.await(); MockNM amNodeManager = rm.registerNode("amNM:1234", 2048); amNodeManager.nodeHeartbeat(true); dispatcher.await(); ApplicationAttemptId appAttemptId = app.getCurrentAppAttempt() .getAppAttemptId(); rm.sendAMLaunched(appAttemptId); dispatcher.await(); JobId jobId = MRBuilderUtils.newJobId(appAttemptId.getApplicationId(), 0); Job mockJob = mock(Job.class); when(mockJob.getReport()).thenReturn( MRBuilderUtils.newJobReport(jobId, "job", "user", JobState.RUNNING, 0, 0, 0, 0, 0, 0, 0, "jobfile", null, false, "")); MyContainerAllocator allocator = new MyContainerAllocator(rm, conf, appAttemptId, mockJob); MockNM nodeManager1 = rm.registerNode("h1:1234", 3072); rm.registerNode("h2:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 1536); dispatcher.await(); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 1024, new String[] { "h1" }); allocator.sendRequest(event2); ContainerRequestEvent event3 = createReq(jobId, 3, 1024, new String[] { "h2" }); allocator.sendRequest(event3); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); nodeManager3.nodeHeartbeat(true); nodeManager1.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); checkAssignments(new ContainerRequestEvent[] { event1, event2, event3 }, assigned, false); for(TaskAttemptContainerAssignedEvent event : assigned) { if(event.getTaskAttemptID().equals(event3.getAttemptID())) { assigned.remove(event); Assert.assertTrue( event.getContainer().getNodeId().getHost().equals("h3")); break; } } checkAssignments(new ContainerRequestEvent[] { event1, event2}, assigned, true); }
private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; } }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; } RMContainerAllocator(ClientService clientService, AppContext context); }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; } RMContainerAllocator(ClientService clientService, AppContext context); boolean getIsReduceStarted(); void setIsReduceStarted(boolean reduceStarted); @Override void handle(ContainerAllocatorEvent event); @Private void scheduleReduces(
      int totalMaps, int completedMaps,
      int scheduledMaps, int scheduledReduces,
      int assignedMaps, int assignedReduces,
      Resource mapResourceReqt, Resource reduceResourceReqt,
      int numPendingReduces,
      float maxReduceRampupLimit, float reduceSlowStart); @Private void scheduleAllReduces(); @Private void rampUpReduces(int rampUp); @Private void rampDownReduces(int rampDown); @VisibleForTesting TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptID); @Private Resource getResourceLimit(); }
RMContainerAllocator extends RMContainerRequestor implements ContainerAllocator { private static String getHost(String contMgrAddress) { String host = contMgrAddress; String[] hostport = host.split(":"); if (hostport.length == 2) { host = hostport[0]; } return host; } RMContainerAllocator(ClientService clientService, AppContext context); boolean getIsReduceStarted(); void setIsReduceStarted(boolean reduceStarted); @Override void handle(ContainerAllocatorEvent event); @Private void scheduleReduces(
      int totalMaps, int completedMaps,
      int scheduledMaps, int scheduledReduces,
      int assignedMaps, int assignedReduces,
      Resource mapResourceReqt, Resource reduceResourceReqt,
      int numPendingReduces,
      float maxReduceRampupLimit, float reduceSlowStart); @Private void scheduleAllReduces(); @Private void rampUpReduces(int rampUp); @Private void rampDownReduces(int rampDown); @VisibleForTesting TaskAttemptEvent createContainerFinishedEvent(ContainerStatus cont,
      TaskAttemptId attemptID); @Private Resource getResourceLimit(); static final float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART; @VisibleForTesting
  static final String RAMPDOWN_DIAGNOSTIC; }