supervisorsByTag.java

private Map<String, ArrayList<SupervisorDetails>> getSupervisorsByTag(
        Collection<SupervisorDetails> supervisorDetails
) {
    // A map of tag -> supervisors, to help with scheduling of components with specific tags
    Map<String, ArrayList<SupervisorDetails>> supervisorsByTag = new HashMap<String, ArrayList<SupervisorDetails>>();

    for (SupervisorDetails supervisor : supervisorDetails) {
        @SuppressWarnings("unchecked")
        Map<String, String> metadata = (Map<String, String>) supervisor.getSchedulerMeta();

        String tags;

        if (metadata == null) {
            tags = "untagged";
        } else {
            tags = metadata.get("tags");

            if (tags == null) {
                tags = "untagged";
            }
        }

        // If the supervisor has tags attached to it, handle it by populating the supervisorsByTag map.
        // Loop through each of the tags to handle individually
        for (String tag : tags.split(",")) {
            tag = tag.trim();

            if (supervisorsByTag.containsKey(tag)) {
                // If we've already seen this tag, then just add the supervisor to the existing ArrayList.
                supervisorsByTag.get(tag).add(supervisor);
            } else {
                // If this tag is new, then create a new ArrayList<SupervisorDetails>,
                // add the current supervisor, and populate the map's tag entry with it.
                ArrayList<SupervisorDetails> newSupervisorList = new ArrayList<SupervisorDetails>();
                newSupervisorList.add(supervisor);
                supervisorsByTag.put(tag, newSupervisorList);
            }
        }
    }

    return supervisorsByTag;
}