[STORM-2931] defaultSchedule function of DefaultScheduler looks not right

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Assignee: Unassigned
• Reporter: Ethan Li
• Created: 02/Feb/18 16:32
• Updated: 02/Feb/18 16:32

Description

I am not sure if this is a bug but the function logic looks not right to me:

DefaultScheduler.java: defaultSchedule()

 

public static void defaultSchedule(Topologies topologies, Cluster cluster) {
    for (TopologyDetails topology : cluster.needsSchedulingTopologies()) {
        List<WorkerSlot> availableSlots = cluster.getAvailableSlots();
        Set<ExecutorDetails> allExecutors = topology.getExecutors();

        Map<WorkerSlot, List<ExecutorDetails>> aliveAssigned = EvenScheduler.getAliveAssignedWorkerSlotExecutors(cluster, topology.getId());
        Set<ExecutorDetails> aliveExecutors = new HashSet<ExecutorDetails>();
        for (List<ExecutorDetails> list : aliveAssigned.values()) {
            aliveExecutors.addAll(list);
        }

        Set<WorkerSlot> canReassignSlots = slotsCanReassign(cluster, aliveAssigned.keySet());
        int totalSlotsToUse = Math.min(topology.getNumWorkers(), canReassignSlots.size() + availableSlots.size());

        Set<WorkerSlot> badSlots = null;
        if (totalSlotsToUse > aliveAssigned.size() || !allExecutors.equals(aliveExecutors)) {
            badSlots = badSlots(aliveAssigned, allExecutors.size(), totalSlotsToUse);                
        }
        if (badSlots != null) {
            cluster.freeSlots(badSlots);
        }

        EvenScheduler.scheduleTopologiesEvenly(new Topologies(topology), cluster);
    }
}

 

EvenScheduler.scheduleTopologiesEvenly:

 

public static void scheduleTopologiesEvenly(Topologies topologies, Cluster cluster) {
    for (TopologyDetails topology : cluster.needsSchedulingTopologies()) {
        String topologyId = topology.getId();
        Map<ExecutorDetails, WorkerSlot> newAssignment = scheduleTopology(topology, cluster);
        Map<WorkerSlot, List<ExecutorDetails>> nodePortToExecutors = Utils.reverseMap(newAssignment);

        for (Map.Entry<WorkerSlot, List<ExecutorDetails>> entry : nodePortToExecutors.entrySet()) {
            WorkerSlot nodePort = entry.getKey();
            List<ExecutorDetails> executors = entry.getValue();
            cluster.assign(nodePort, topologyId, executors);
        }
    }
}

 

In the for-loop of DefaultScheduler.defaultSchedule(), it calls EvenScheduler.scheduleTopologiesEvenly()  every time. But  there is another for-loop in EvenScheduler.scheduleTopologiesEvenly(). These two for-loops both iterate on 

cluster.needsSchedulingTopologies() 

and never uses the first parameter which is 

Topologies topologies