@Test public void testBlackListedNodes() throws Exception { LOG.info("Running testBlackListedNodes"); Configuration conf = new Configuration(); conf.setBoolean(MRJobConfig.MR_AM_JOB_NODE_BLACKLISTING_ENABLE, true); conf.setInt(MRJobConfig.MAX_TASK_FAILURES_PER_TRACKER, 1); conf.setInt( MRJobConfig.MR_AM_IGNORE_BLACKLISTING_BLACKLISTED_NODE_PERECENT, -1); 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", 10240); MockNM nodeManager2 = rm.registerNode("h2:1234", 10240); MockNM nodeManager3 = rm.registerNode("h3:1234", 10240); dispatcher.await(); ContainerRequestEvent event1 = createReq(jobId, 1, 1024, new String[] { "h1" }); allocator.sendRequest(event1); ContainerRequestEvent event2 = createReq(jobId, 2, 1024, new String[] { "h2" }); allocator.sendRequest(event2); ContainerRequestEvent event3 = createReq(jobId, 3, 1024, new String[] { "h3" }); allocator.sendRequest(event3); List<TaskAttemptContainerAssignedEvent> assigned = allocator.schedule(); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); ContainerFailedEvent f1 = createFailEvent(jobId, 1, "h1", false); allocator.sendFailure(f1); ContainerFailedEvent f2 = createFailEvent(jobId, 1, "h2", false); allocator.sendFailure(f2); nodeManager1.nodeHeartbeat(true); nodeManager2.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); dispatcher.await(); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); assertBlacklistAdditionsAndRemovals(2, 0, rm); nodeManager1.nodeHeartbeat(false); nodeManager2.nodeHeartbeat(false); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertEquals("No of assignments must be 0", 0, assigned.size()); nodeManager3.nodeHeartbeat(true); dispatcher.await(); assigned = allocator.schedule(); dispatcher.await(); assertBlacklistAdditionsAndRemovals(0, 0, rm); Assert.assertTrue("No of assignments must be 3", assigned.size() == 3); for (TaskAttemptContainerAssignedEvent assig : assigned) { Assert.assertTrue("Assigned container host not correct", "h3".equals(assig .getContainer().getNodeId().getHost())); } }
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(); @VisibleForTesting int getNumOfPendingReduces(); }
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(); @VisibleForTesting int getNumOfPendingReduces(); static final float DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART; @VisibleForTesting
  static final String RAMPDOWN_DIAGNOSTIC; }