public static VerifiableProducer createFromArgs(ArgumentParser parser, String[] args) throws ArgumentParserException {
        Namespace res = parser.parseArgs(args);

        int maxMessages = res.getInt("maxMessages");
        String topic = res.getString("topic");
        int throughput = res.getInt("throughput");
        String configFile = res.getString("producer.config");
        Integer valuePrefix = res.getInt("valuePrefix");
        Long createTime = res.getLong("createTime");
        Integer repeatingKeys = res.getInt("repeatingKeys");

        if (createTime == -1L)
            createTime = null;

        Properties producerProps = new Properties();

        if (res.get("bootstrapServer") != null) {
            producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, res.getString("bootstrapServer"));
        } else if (res.getString("brokerList") != null) {
            producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, res.getString("brokerList"));
        } else {
            parser.printHelp();
            // Can't use `Exit.exit` here because it didn't exist until 0.11.0.0.
            System.exit(0);
        }

        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        producerProps.put(ProducerConfig.ACKS_CONFIG, Integer.toString(res.getInt("acks")));
        // No producer retries
        producerProps.put(ProducerConfig.RETRIES_CONFIG, "0");
        if (configFile != null) {
            try {
                producerProps.putAll(loadProps(configFile));
            } catch (IOException e) {
                throw new ArgumentParserException(e.getMessage(), parser);
            }
        }

        StringSerializer serializer = new StringSerializer();
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps, serializer, serializer);

        return new VerifiableProducer(producer, topic, throughput, maxMessages, valuePrefix, createTime, repeatingKeys);
    }
Construct a VerifiableProducer object from command-line arguments.
tools/src/main/java/org/apache/kafka/tools/VerifiableProducer.java