tagAwareSchedule.java

private void tagAwareSchedule(Topologies topologies, Cluster cluster) {
    Collection<SupervisorDetails> supervisorDetails = cluster.getSupervisors().values();

    // Get the lists of tagged and unreserved supervisors.
    Map<String, ArrayList<SupervisorDetails>> supervisorsByTag = getSupervisorsByTag(supervisorDetails);

    for (TopologyDetails topologyDetails : cluster.needsSchedulingTopologies(topologies)) {
        StormTopology stormTopology = topologyDetails.getTopology();
        String topologyID = topologyDetails.getId();

        // Get components from topology
        Map<String, Bolt> bolts = stormTopology.get_bolts();
        Map<String, SpoutSpec> spouts = stormTopology.get_spouts();

        // Get a map of component to executors
        Map<String, List<ExecutorDetails>> executorsByComponent = cluster.getNeedsSchedulingComponentToExecutors(
                topologyDetails
        );

        // Get a map of tag to components
        Map<String, ArrayList<String>> componentsByTag = new HashMap<String, ArrayList<String>>();
        populateComponentsByTag(componentsByTag, bolts);
        populateComponentsByTag(componentsByTag, spouts);
        populateComponentsByTagWithStormInternals(componentsByTag, executorsByComponent.keySet());
    }
}