DefaultScheduler implements IScheduler, IMesosStormScheduler { @Override public void schedule(Topologies topologies, Cluster cluster) { List<WorkerSlot> workerSlots = cluster.getAvailableSlots(); Map<String, List<MesosWorkerSlot>> perTopologySlotList = getMesosWorkerSlotPerTopology(workerSlots); for (String topologyId : perTopologySlotList.keySet()) { TopologyDetails topologyDetails = topologies.getById(topologyId); List<MesosWorkerSlot> mesosWorkerSlots = perTopologySlotList.get(topologyId); int countSlotsRequested = topologyDetails.getNumWorkers(); int countSlotsAssigned = cluster.getAssignedNumWorkers(topologyDetails); if (mesosWorkerSlots.size() == 0) { log.warn("No slots found for topology {} while scheduling", topologyId); continue; } int countSlotsAvailable = Math.min(mesosWorkerSlots.size(), (countSlotsRequested - countSlotsAssigned)); List<List<ExecutorDetails>> executorsPerWorkerList = executorsPerWorkerList(cluster, topologyDetails, countSlotsAvailable); for (int i = 0; i < countSlotsAvailable; i++) { cluster.assign(mesosWorkerSlots.remove(0), topologyId, executorsPerWorkerList.remove(0)); } } mesosWorkerSlotMap.clear(); }  @Override void prepare(Map conf); @Override List<WorkerSlot> allSlotsAvailableForScheduling(RotatingMap<Protos.OfferID, Protos.Offer> offers,
                                                         Collection<SupervisorDetails> existingSupervisors,
                                                         Topologies topologies, Set<String> topologiesMissingAssignments); @Override void schedule(Topologies topologies, Cluster cluster);  }
@Test public void testScheduleWithOneWorkerSlot() { Cluster spyCluster = getSpyCluster(); List<MesosWorkerSlot> mesosWorkerSlots = this.generateMesosWorkerSlots(1); initializeMesosWorkerSlotMap(mesosWorkerSlots); Set<ExecutorDetails> executorsToAssign = this.generateExecutorDetailsSet(4); List<WorkerSlot> workerSlotList = getWorkerSlotFromMesosWorkerSlot(mesosWorkerSlots); doReturn(workerSlotList).when(spyCluster).getAvailableSlots(); doReturn(executorsToAssign).when(spyCluster).getUnassignedExecutors(any(TopologyDetails.class)); defaultScheduler.schedule(topologies, spyCluster); Set<ExecutorDetails> assignedExecutors = spyCluster.getAssignmentById(sampleTopologyId).getExecutors(); assertEquals(executorsToAssign, assignedExecutors); }
@Test public void testScheduleWithMultipleSlotsOnSameHost() { Cluster spyCluster = this.getSpyCluster(3, 3); defaultScheduler.schedule(topologies, spyCluster); SchedulerAssignment schedulerAssignment = spyCluster.getAssignments() .get(sampleTopologyId); Map<ExecutorDetails, WorkerSlot> executorDetailsWorkerSlotMap = schedulerAssignment.getExecutorToSlot(); assertEquals(executorDetailsWorkerSlotMap.keySet().size(), 3); assertEquals(executorDetailsWorkerSlotMap.values().size(), 3); spyCluster = this.getSpyCluster(3, 6); defaultScheduler.schedule(topologies, spyCluster); executorDetailsWorkerSlotMap = spyCluster.getAssignments() .get(sampleTopologyId) .getExecutorToSlot(); assertEquals(executorDetailsWorkerSlotMap.keySet().size(), 6); int workerSlotsUsed = new HashSet<>(executorDetailsWorkerSlotMap.values()).size() ; assertEquals(workerSlotsUsed, 3); schedulerAssignment = spyCluster.getAssignments() .get(sampleTopologyId); executorDetailsWorkerSlotMap = schedulerAssignment.getExecutorToSlot(); Map<WorkerSlot, List<ExecutorDetails>> workerSlotExecutorDetailsMap = this.getworkerSlotExecutorDetailsMap(executorDetailsWorkerSlotMap); for (WorkerSlot workerSlot : workerSlotExecutorDetailsMap.keySet()) { List<ExecutorDetails> executorDetails = workerSlotExecutorDetailsMap.get(workerSlot); assertEquals(3, Math.abs(executorDetails.get(0).getStartTask() - executorDetails.get(1).getEndTask())); } }