NMHeartBeatHandler extends BaseInterceptor { @VisibleForTesting protected Resource getNewResourcesOfferedByMesos(String hostname) { OfferFeed feed = offerLifecycleMgr.getOfferFeed(hostname); List<Offer> offers = new ArrayList<>(); Protos.Offer offer; while ((offer = feed.poll()) != null) { offers.add(offer); offerLifecycleMgr.markAsConsumed(offer); } Resource fromMesosOffers = OfferUtils.getYarnResourcesFromMesosOffers(offers); if (logger.isDebugEnabled()) { logger.debug("NM on host {} got {} CPUs and {} memory from mesos", hostname, fromMesosOffers.getVirtualCores(), fromMesosOffers.getMemory()); } return fromMesosOffers; } @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 testGetNewResourcesOfferedByMesos() throws Exception { 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", 2.0, 1024.0); olManager.addOffers(offerOne); olManager.addOffers(offerTwo); Resource resourcesOne = handler.getNewResourcesOfferedByMesos("localhost-one"); assertEquals(1.0, resourcesOne.getVirtualCores(), 0.0); assertEquals(512.0, resourcesOne.getMemory(), 0.0); Resource resourcesTwo = handler.getNewResourcesOfferedByMesos("localhost-two"); assertEquals(2.0, resourcesTwo.getVirtualCores(), 0.0); assertEquals(1024.0, resourcesTwo.getMemory(), 0.0); }