private static void validateKafkaConfig(List<ImportConfigurationType> configs) {
        if (configs.isEmpty()) {
            return;
        }
        // We associate each group id with the set of topics that belong to it
        HashMap<String, HashSet<String>> groupidToTopics = new HashMap<>();
        for (ImportConfigurationType config : configs) {
            String groupid = "";
            HashSet<String> topics = new HashSet<>();
            // Fetch topics and group id from each configuration
            for (PropertyType pt : config.getProperty()) {
                if (pt.getName().equals("topics")) {
                    topics.addAll(Arrays.asList(pt.getValue().split("\\s*,\\s*")));
                } else if (pt.getName().equals("groupid")) {
                    groupid = pt.getValue();
                }
            }
            if (groupidToTopics.containsKey(groupid)) {
                // Under this group id, we first union the set of already-stored topics with the set of newly-seen topics.
                HashSet<String> union = new HashSet<>(groupidToTopics.get(groupid));
                union.addAll(topics);
                if (union.size() == (topics.size() + groupidToTopics.get(groupid).size())) {
                    groupidToTopics.put(groupid, union);
                } else {
                    // If the size of the union doesn't equal to the sum of sizes of newly-seen topic set and
                    // already-stored topic set, those two sets must overlap with each other, which means that
                    // there must be two configurations having the same group id and overlapping sets of topics.
                    // Thus, we throw the RuntimeException.
                    throw new RuntimeException("Invalid import configuration. Two Kafka entries have the same groupid and topic.");
                }
            } else {
                groupidToTopics.put(groupid, topics);
            }
        }
    }
private static void validateKafkaConfig ( List < ImportConfigurationType > configs ) { if ( configs . isEmpty ( ) ) { return ; } HashMap < String , HashSet < String > > groupidToTopics = new HashMap < > ( ) ; for ( ImportConfigurationType config : configs ) { String groupid = "" ; HashSet < String > topics = new HashSet < > ( ) ; for ( PropertyType pt : config . getProperty ( ) ) { if ( pt . getName ( ) . equals ( "topics" ) ) { topics . addAll ( Arrays . asList ( pt . getValue ( ) . split ( "\\s*,\\s*" ) ) ) ; } else if ( pt . getName ( ) . equals ( "groupid" ) ) { groupid = pt . getValue ( ) ; } } if ( groupidToTopics . containsKey ( groupid ) ) { HashSet < String > union = new HashSet < > ( groupidToTopics . get ( groupid ) ) ; union . addAll ( topics ) ; if ( union . size ( ) == ( topics . size ( ) + groupidToTopics . get ( groupid ) . size ( ) ) ) { groupidToTopics . put ( groupid , union ) ; } else { throw new RuntimeException ( "Invalid import configuration. Two Kafka entries have the same groupid and topic." ) ; } } else { groupidToTopics . put ( groupid , topics ) ; } } }
Check whether two Kafka configurations have both the same topic and group id. If two configurations
have the same group id and overlapping sets of topics, a RuntimeException will be thrown.
@param configs All parsed Kafka configurations