private static void mergeKafka10ImportConfigurations(Map<String, ImportConfiguration> processorConfig) {
        if (processorConfig.isEmpty()) {
            return;
        }

        Map<String, ImportConfiguration> kafka10ProcessorConfigs = new HashMap<>();
        Iterator<Map.Entry<String, ImportConfiguration>> iter = processorConfig.entrySet().iterator();
        while (iter.hasNext()) {
            String configName = iter.next().getKey();
            ImportConfiguration importConfig = processorConfig.get(configName);
            Properties properties = importConfig.getmoduleProperties();

            String importBundleJar = properties.getProperty(ImportDataProcessor.IMPORT_MODULE);
            Preconditions.checkNotNull(importBundleJar,
                    "Import source is undefined or custom import plugin class missing.");
            //handle special cases for kafka 10 and maybe late versions
            String[] bundleJar = importBundleJar.split("kafkastream");
            if (bundleJar.length > 1) {
                String version = bundleJar[1].substring(0, bundleJar[1].indexOf(".jar"));
                if (!version.isEmpty()) {
                    int versionNumber = Integer.parseInt(version);
                    if (versionNumber == 10) {
                        kafka10ProcessorConfigs.put(configName, importConfig);
                        iter.remove();
                    }
                }
            }
        }

        if (kafka10ProcessorConfigs.isEmpty()) {
            return;
        }

        Map<String, ImportConfiguration> mergedConfigs = new HashMap<>();
        iter = kafka10ProcessorConfigs.entrySet().iterator();
        while (iter.hasNext()) {
            ImportConfiguration importConfig = iter.next().getValue();
            Properties props = importConfig.getmoduleProperties();

            //organize the kafka10 importer by the broker list and group id
            //All importers must be configured by either broker list or zookeeper in the same group
            //otherwise, these importers can not be correctly merged.
            String brokers = props.getProperty("brokers");
            String groupid = props.getProperty("groupid", "voltdb");
            if (brokers == null) {
                brokers = props.getProperty("zookeeper");
            }
            String brokersGroup = brokers + "_" + groupid;
            ImportConfiguration config = mergedConfigs.get(brokersGroup);
            if (config == null) {
                mergedConfigs.put(brokersGroup, importConfig);
            } else {
                config.mergeProperties(props);
            }
        }
        processorConfig.putAll(mergedConfigs);
    }
private static void mergeKafka10ImportConfigurations ( Map < String , ImportConfiguration > processorConfig ) { if ( processorConfig . isEmpty ( ) ) { return ; } Map < String , ImportConfiguration > kafka10ProcessorConfigs = new HashMap < > ( ) ; Iterator < Map . Entry < String , ImportConfiguration > > iter = processorConfig . entrySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { String configName = iter . next ( ) . getKey ( ) ; ImportConfiguration importConfig = processorConfig . get ( configName ) ; Properties properties = importConfig . getmoduleProperties ( ) ; String importBundleJar = properties . getProperty ( ImportDataProcessor . IMPORT_MODULE ) ; Preconditions . checkNotNull ( importBundleJar , "Import source is undefined or custom import plugin class missing." ) ; String [ ] bundleJar = importBundleJar . split ( "kafkastream" ) ; if ( bundleJar . length > 1 ) { String version = bundleJar [ 1 ] . substring ( 0 , bundleJar [ 1 ] . indexOf ( ".jar" ) ) ; if ( ! version . isEmpty ( ) ) { int versionNumber = Integer . parseInt ( version ) ; if ( versionNumber == 10 ) { kafka10ProcessorConfigs . put ( configName , importConfig ) ; iter . remove ( ) ; } } } } if ( kafka10ProcessorConfigs . isEmpty ( ) ) { return ; } Map < String , ImportConfiguration > mergedConfigs = new HashMap < > ( ) ; iter = kafka10ProcessorConfigs . entrySet ( ) . iterator ( ) ; while ( iter . hasNext ( ) ) { ImportConfiguration importConfig = iter . next ( ) . getValue ( ) ; Properties props = importConfig . getmoduleProperties ( ) ; String brokers = props . getProperty ( "brokers" ) ; String groupid = props . getProperty ( "groupid" , "voltdb" ) ; if ( brokers == null ) { brokers = props . getProperty ( "zookeeper" ) ; } String brokersGroup = brokers + "_" + groupid ; ImportConfiguration config = mergedConfigs . get ( brokersGroup ) ; if ( config == null ) { mergedConfigs . put ( brokersGroup , importConfig ) ; } else { config . mergeProperties ( props ) ; } } processorConfig . putAll ( mergedConfigs ) ; }
aggregate Kafka10 importer configurations.One importer per brokers and kafka group. Formatters and stored procedures
can vary by topics.