class 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;
    }

}