public static VerifiableLog4jAppender createFromArgs(String[] args) {
        ArgumentParser parser = argParser();
        VerifiableLog4jAppender producer = null;

        try {
            Namespace res = parser.parseArgs(args);

            int maxMessages = res.getInt("maxMessages");
            String topic = res.getString("topic");
            String configFile = res.getString("appender.config");

            Properties props = new Properties();
            props.setProperty("log4j.rootLogger", "INFO, KAFKA");
            props.setProperty("log4j.appender.KAFKA", "org.apache.kafka.log4jappender.KafkaLog4jAppender");
            props.setProperty("log4j.appender.KAFKA.layout", "org.apache.log4j.PatternLayout");
            props.setProperty("log4j.appender.KAFKA.layout.ConversionPattern", "%-5p: %c - %m%n");
            props.setProperty("log4j.appender.KAFKA.BrokerList", res.getString("brokerList"));
            props.setProperty("log4j.appender.KAFKA.Topic", topic);
            props.setProperty("log4j.appender.KAFKA.RequiredNumAcks", res.getString("acks"));
            props.setProperty("log4j.appender.KAFKA.SyncSend", "true");
            final String securityProtocol = res.getString("securityProtocol");
            if (securityProtocol != null && !securityProtocol.equals(SecurityProtocol.PLAINTEXT.toString())) {
                props.setProperty("log4j.appender.KAFKA.SecurityProtocol", securityProtocol);
            }
            if (securityProtocol != null && securityProtocol.contains("SSL")) {
                props.setProperty("log4j.appender.KAFKA.SslTruststoreLocation", res.getString("sslTruststoreLocation"));
                props.setProperty("log4j.appender.KAFKA.SslTruststorePassword", res.getString("sslTruststorePassword"));
            }
            if (securityProtocol != null && securityProtocol.contains("SASL")) {
                props.setProperty("log4j.appender.KAFKA.SaslKerberosServiceName", res.getString("saslKerberosServiceName"));
                props.setProperty("log4j.appender.KAFKA.clientJaasConfPath", res.getString("clientJaasConfPath"));
                props.setProperty("log4j.appender.KAFKA.kerb5ConfPath", res.getString("kerb5ConfPath"));
            }
            props.setProperty("log4j.logger.kafka.log4j", "INFO, KAFKA");
            // Changing log level from INFO to WARN as a temporary workaround for KAFKA-6415. This is to
            // avoid deadlock in system tests when producer network thread appends to log while updating metadata.
            props.setProperty("log4j.logger.org.apache.kafka.clients.Metadata", "WARN, KAFKA");

            if (configFile != null) {
                try {
                    props.putAll(loadProps(configFile));
                } catch (IOException e) {
                    throw new ArgumentParserException(e.getMessage(), parser);
                }
            }

            producer = new VerifiableLog4jAppender(props, maxMessages);
        } catch (ArgumentParserException e) {
            if (args.length == 0) {
                parser.printHelp();
                Exit.exit(0);
            } else {
                parser.handleError(e);
                Exit.exit(1);
            }
        }

        return producer;
    }
Construct a VerifiableLog4jAppender object from command-line arguments.
tools/src/main/java/org/apache/kafka/tools/VerifiableLog4jAppender.java