SchedulerUtils { public static Map<String, List<OfferResources>> getOfferResourcesListPerNode(RotatingMap<Protos.OfferID, Protos.Offer> offers) { Map<String, List<OfferResources>> offerResourcesListPerNode = new HashMap<>(); for (Protos.Offer offer : offers.values()) { String hostName = offer.getHostname(); List<OfferResources> offerResourcesListForCurrentHost = offerResourcesListPerNode.get(hostName); OfferResources offerResources = new OfferResources(offer); if (offerResourcesListForCurrentHost == null) { offerResourcesListPerNode.put(hostName, new ArrayList<OfferResources>()); } offerResourcesListPerNode.get(hostName).add(offerResources); log.info("Available resources at {}: {}", hostName, offerResources.toString()); } return offerResourcesListPerNode; }  static boolean isFit(Map mesosStormConf, OfferResources offerResources, TopologyDetails topologyDetails, boolean supervisorExists); static boolean isFit(Map mesosStormConf, Protos.Offer offer, TopologyDetails topologyDetails, boolean supervisorExists); static boolean supervisorExists(String offerHost, Collection<SupervisorDetails> existingSupervisors,
                                   String topologyId); static Map<String, List<OfferResources>> getOfferResourcesListPerNode(RotatingMap<Protos.OfferID, Protos.Offer> offers); static MesosWorkerSlot createWorkerSlotFromOfferResources(Map mesosStormConf, OfferResources offerResources,
                                                            TopologyDetails topologyDetails, boolean supervisorExists);  }
@Test public void testGetOfferResourcesListPerNode() { String hostName = sampleHost; buildOfferAndUpdateRotatingMap("offer1", hostName, 0, 1000); buildOfferAndUpdateRotatingMap("offer2", hostName, 10, 0); buildOfferAndUpdateRotatingMap("offer3", hostName, 0, 100.01); buildOfferAndUpdateRotatingMap("offer4", hostName, 1.001, 0); buildOfferAndUpdateRotatingMap("offer5", hostName, 0, 0.001); buildOfferAndUpdateRotatingMap("offer6", hostName, 0.001, 0.01); Map<String, List<OfferResources>> offerResourcesMap = SchedulerUtils.getOfferResourcesListPerNode(rotatingMap); assertEquals(offerResourcesMap.size(), 1); List<OfferResources> offerResources = offerResourcesMap.get("host1.east"); assertEquals(offerResources.size(), 6); hostName = "host1.west"; buildOfferAndUpdateRotatingMap("offer7", hostName, 0, 1000); buildOfferAndUpdateRotatingMap("offer8", hostName, 10, 0); offerResourcesMap = SchedulerUtils.getOfferResourcesListPerNode(rotatingMap); assertEquals(offerResourcesMap.size(), 2); offerResources = offerResourcesMap.get("host1.east"); assertEquals(offerResources.size(), 6); offerResources = offerResourcesMap.get("host1.west"); assertEquals(offerResources.size(), 2); }