protected Map<String, String> readLog4jConfig(String config) {
        Map<String, String> parsed = new LinkedHashMap<>();
        Map<String, String> env = new HashMap<>();
        BufferedReader firstPassReader = new BufferedReader(new StringReader(config));
        BufferedReader reader = new BufferedReader(new StringReader(config));
        String line;

        // first pass to lookup the var definitions
        try {
            while ((line = firstPassReader.readLine()) != null) {
                // skip comments
                if (line.startsWith("#")) continue;

                // ignore empty lines
                line = line.trim();
                if (line.length() == 0) continue;

                // everything that does not start with 'log4j.' is a variable definition
                if (!line.startsWith("log4j.")) {
                    int foundIdx = line.indexOf("=");
                    if (foundIdx >= 0) {
                        env.put(line.substring(0, foundIdx).trim(), line.substring(foundIdx + 1).trim());
                    } else {
                        env.put(line.trim(), "");
                    }
                    LOGGER.debugCr(reconciliation, "Treating the line as ENV var declaration: {}", line);
                }
            }
        } catch (Exception e) {
            LOGGER.errorCr(reconciliation, "Failed to parse logging configuration: " + config, e);
            return Collections.emptyMap();
        }

        try {
            while ((line = reader.readLine()) != null) {
                // skip comments
                if (line.startsWith("#")) continue;

                // ignore empty lines
                line = line.trim();
                if (line.length() == 0) continue;

                // we ignore appenders (log4j.appender.*)
                // and only handle loggers (log4j.logger.*)
                if (line.startsWith("log4j.logger.")) {
                    int startIdx = "log4j.logger.".length();
                    int endIdx = line.indexOf("=", startIdx);
                    if (endIdx == -1) {
                        LOGGER.debugCr(reconciliation, "Skipping log4j.logger.* declaration without level: {}", line);
                        continue;
                    }
                    String name = line.substring(startIdx, endIdx).trim();
                    String value = line.substring(endIdx + 1).split(",")[0].trim();

                    value = Util.expandVar(value, env);
                    parsed.put(name, value);

                } else if (line.startsWith("log4j.rootLogger=")) {
                    int startIdx = "log4j.rootLogger=".length();
                    parsed.put("root", Util.expandVar(line.substring(startIdx).split(",")[0].trim(), env));

                } else {
                    LOGGER.debugCr(reconciliation, "Skipping log4j line: {}", line);
                }
            }
        } catch (Exception e) {
            LOGGER.errorCr(reconciliation, "Failed to parse logging configuration: " + config, e);
            return Collections.emptyMap();
        }
        return parsed;
    }
@Test
    public void testExpansion() {
        String input = "log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender\n" +
                "log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout\n" +
                "log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} %p %m (%c) [%t]%n\n" +
                "log4j.rootLogger=${kafka.root.logger.level}, CONSOLE\n" +
                "\n" +
                "# Log levels\n" +
                // level definition is afterwards its use
                "kafka.root.logger.level=INFO\n" +
                "log4j.logger.org.apache.zookeeper=INFO\n" +
                "log4j.logger.kafka=INFO\n" +
                "log4j.logger.org.apache.kafka=INFO\n" +
                "log4j.logger.kafka.request.logger=WARN, CONSOLE\n" +
                "log4j.logger.kafka.network.Processor=OFF\n" +
                "log4j.logger.kafka.server.KafkaApis=OFF\n" +
                "log4j.logger.kafka.network.RequestChannel$=WARN\n" +
                "log4j.logger.kafka.controller=TRACE\n" +
                "log4j.logger.kafka.log.LogCleaner=INFO\n" +
                "log4j.logger.state.change.logger=TRACE\n" +
                "log4j.logger.kafka.authorizer.logger=INFO\n" +
                "monitorInterval=30\n";

        KafkaBrokerLoggingConfigurationDiff kdiff = new KafkaBrokerLoggingConfigurationDiff(Reconciliation.DUMMY_RECONCILIATION, null, null);

        Map<String, String> res = kdiff.readLog4jConfig(input);
        assertThat(res.get("root"), is("INFO"));
        assertThat(res.get("kafka.request.logger"), is("WARN"));
    }