KafkaConfigurationParser { static List<KafkaConfiguration> parse(Buffer configurationResourceBuffer, Map<String, Object> properties) { String replacedConfig; JsonObject config; List<KafkaConfiguration> configurations = new ArrayList<>(); try { replacedConfig = StringUtils.replaceWildcardConfigs(configurationResourceBuffer.toString(UTF_8), properties); config = new JsonObject(Buffer.buffer(replacedConfig)); } catch (Exception e) { log.warn("Could not replace wildcards with environment properties for kafka configurations due to following reason: {}", e.getMessage()); return configurations; } for (String topicPattern : config.fieldNames()) { try { Pattern pattern = Pattern.compile(topicPattern); final Map<String, String> additionalConfig = extractAdditionalConfig(config.getJsonObject(topicPattern)); configurations.add(new KafkaConfiguration(pattern, additionalConfig)); log.info("Topic '{}' successfully parsed and added to kafka configuration list", topicPattern); } catch (PatternSyntaxException patternException) { log.warn("Topic '{}' is not a valid regex pattern. Discarding this kafka configuration", topicPattern); } } return configurations; }    }
@Test public void parseAllValid(TestContext context) { List<KafkaConfiguration> configurations = KafkaConfigurationParser.parse(Buffer.buffer(CONFIG_RESOURCE), new HashMap<>()); context.assertEquals(2, configurations.size()); KafkaConfiguration config_1 = configurations.get(0); context.assertEquals(Pattern.compile("my.topic.*").pattern(), config_1.getTopic().pattern()); Map<String, String> expected_1 = new HashMap<String, String>() {{ put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("bootstrap.servers", "localhost:9092"); }}; context.assertEquals(expected_1, config_1.getConfigurations()); KafkaConfiguration config_2 = configurations.get(1); context.assertEquals(Pattern.compile(".").pattern(), config_2.getTopic().pattern()); Map<String, String> expected_2 = new HashMap<String, String>() {{ put("bootstrap.servers", "localhost:9093"); put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("acks", "1"); }}; context.assertEquals(expected_2, config_2.getConfigurations()); }
@Test public void parseWithInvalidPattern(TestContext context) { List<KafkaConfiguration> configurations = KafkaConfigurationParser.parse(Buffer.buffer(INVALID_PATTERN_CONFIG_RESOURCE), new HashMap<>()); context.assertEquals(1, configurations.size()); KafkaConfiguration config_1 = configurations.get(0); context.assertEquals(Pattern.compile("my.topic.*").pattern(), config_1.getTopic().pattern()); Map<String, String> expected_1 = new HashMap<String, String>() {{ put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("bootstrap.servers", "localhost:9092"); }}; context.assertEquals(expected_1, config_1.getConfigurations()); }
@Test public void parseWildcardValid(TestContext context) { Map<String, Object> properties = new HashMap<>(); properties.put("kafka.host", "localhost"); properties.put("kafka.port", "9095"); List<KafkaConfiguration> configurations = KafkaConfigurationParser.parse(Buffer.buffer(CONFIG_WILDCARD_RESOURCE), properties); context.assertEquals(1, configurations.size()); KafkaConfiguration config = configurations.get(0); context.assertEquals(Pattern.compile("my.properties.topic.*").pattern(), config.getTopic().pattern()); Map<String, String> expected_1 = new HashMap<String, String>() {{ put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); put("acks", "all"); put("bootstrap.servers", "localhost:9095"); put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); }}; context.assertEquals(expected_1, config.getConfigurations()); }
@Test public void parseWildcardNoEnvProps(TestContext context) { List<KafkaConfiguration> configurations = KafkaConfigurationParser.parse(Buffer.buffer(CONFIG_WILDCARD_RESOURCE), new HashMap<>()); context.assertEquals(0, configurations.size()); }