NMHeartBeatHandler extends BaseInterceptor { @Override public void beforeRMNodeEventHandled(RMNodeEvent event, RMContext context) { switch (event.getType()) { case STARTED: RMNode rmNode = context.getRMNodes().get(event.getNodeId()); if (isNonZeroCapacityNode(rmNode)) { Resource totalCapability = rmNode.getTotalCapability(); logger.warn( "FineGrainedScaling feature got invoked for a NM with non-zero capacity. Host: {}, Mem: {}, CPU: {}. Setting the " + "NM's capacity to (0G,0CPU)", rmNode.getHostName(), totalCapability.getMemory(), totalCapability.getVirtualCores()); totalCapability.setMemory(0); totalCapability.setVirtualCores(0); } break; case STATUS_UPDATE: handleStatusUpdate(event, context); break; default: break; } } @Inject  NMHeartBeatHandler(InterceptorRegistry registry, AbstractYarnScheduler yarnScheduler, MyriadDriver myriadDriver,
                            YarnNodeCapacityManager yarnNodeCapacityMgr, OfferLifecycleManager offerLifecycleMgr,
                            NodeStore nodeStore, SchedulerState state, NodeManagerConfiguration conf); @Override CallBackFilter getCallBackFilter(); @Override void beforeRMNodeEventHandled(RMNodeEvent event, RMContext context);  }
@Test public void testIncrementNodeCapacityUnderCapacity() throws Exception { resetNodeTotalCapability(nodeOne, 0, 0); resetNodeTotalCapability(nodeTwo, 2, 512); Offer offerOne = TestObjectFactory.getOffer("localhost-one", "slave-one", "mock", "offer-one", 1.0, 512.0); Offer offerTwo = TestObjectFactory.getOffer("localhost-two", "slave-two", "mock", "offer-two", 3.0, 1024.0); olManager.addOffers(offerOne); olManager.addOffers(offerTwo); RMNodeStatusEvent eventOne = getRMStatusEvent(nodeOne); handler.beforeRMNodeEventHandled(eventOne, context); RMNodeStatusEvent eventTwo = getRMStatusEvent(nodeTwo); handler.beforeRMNodeEventHandled(eventTwo, context); assertEquals(512, nodeOne.getTotalCapability().getMemory()); assertEquals(1, nodeOne.getTotalCapability().getVirtualCores()); assertEquals(1024, nodeTwo.getTotalCapability().getMemory()); assertEquals(3, nodeTwo.getTotalCapability().getVirtualCores()); }
@Test public void testIncrementNodeCapacityOverCapacity() throws Exception { resetNodeTotalCapability(nodeOne, 1, 512); resetNodeTotalCapability(nodeTwo, 2, 2048); Offer offerOne = TestObjectFactory.getOffer("localhost-one", "slave-one", "mock", "offer-one", 0.2, 3072.0); Offer offerTwo = TestObjectFactory.getOffer("localhost-two", "slave-two", "mock", "offer-two", 8.0, 1024.0); olManager.addOffers(offerOne); olManager.addOffers(offerTwo); RMNodeStatusEvent eventOne = getRMStatusEvent(nodeOne); handler.beforeRMNodeEventHandled(eventOne, context); RMNodeStatusEvent eventTwo = getRMStatusEvent(nodeTwo); handler.beforeRMNodeEventHandled(eventTwo, context); assertEquals(512, nodeOne.getTotalCapability().getMemory()); assertEquals(1, nodeOne.getTotalCapability().getVirtualCores()); assertEquals(2048, nodeTwo.getTotalCapability().getMemory()); assertEquals(2, nodeTwo.getTotalCapability().getVirtualCores()); }