static ArgumentParser argParser() {
        ArgumentParser parser = ArgumentParsers
                .newArgumentParser("producer-performance")
                .defaultHelp(true)
                .description("This tool is used to verify the producer performance. To enable transactions, " +
                        "you can specify a transaction id or set a transaction duration using --transaction-duration-ms. " +
                        "There are three ways to specify the transaction id: set transaction.id=<id> via --producer-props, " +
                        "set transaction.id=<id> in the config file via --producer.config, or use --transaction-id <id>.");

        MutuallyExclusiveGroup payloadOptions = parser
                .addMutuallyExclusiveGroup()
                .required(true)
                .description("either --record-size or --payload-file must be specified but not both.");

        parser.addArgument("--topic")
                .action(store())
                .required(true)
                .type(String.class)
                .metavar("TOPIC")
                .help("produce messages to this topic");

        parser.addArgument("--num-records")
                .action(store())
                .required(true)
                .type(Long.class)
                .metavar("NUM-RECORDS")
                .dest("numRecords")
                .help("number of messages to produce");

        payloadOptions.addArgument("--record-size")
                .action(store())
                .required(false)
                .type(Integer.class)
                .metavar("RECORD-SIZE")
                .dest("recordSize")
                .help("message size in bytes. Note that you must provide exactly one of --record-size or --payload-file " +
                        "or --payload-monotonic.");

        payloadOptions.addArgument("--payload-file")
                .action(store())
                .required(false)
                .type(String.class)
                .metavar("PAYLOAD-FILE")
                .dest("payloadFile")
                .help("file to read the message payloads from. This works only for UTF-8 encoded text files. " +
                        "Payloads will be read from this file and a payload will be randomly selected when sending messages. " +
                        "Note that you must provide exactly one of --record-size or --payload-file or --payload-monotonic.");

        payloadOptions.addArgument("--payload-monotonic")
                .action(storeTrue())
                .type(Boolean.class)
                .metavar("PAYLOAD-MONOTONIC")
                .dest("payloadMonotonic")
                .help("payload is monotonically increasing integer. Note that you must provide exactly one of --record-size " +
                        "or --payload-file or --payload-monotonic.");

        parser.addArgument("--payload-delimiter")
                .action(store())
                .required(false)
                .type(String.class)
                .metavar("PAYLOAD-DELIMITER")
                .dest("payloadDelimiter")
                .setDefault("\\n")
                .help("provides delimiter to be used when --payload-file is provided. " +
                        "Defaults to new line. " +
                        "Note that this parameter will be ignored if --payload-file is not provided.");

        parser.addArgument("--throughput")
                .action(store())
                .required(true)
                .type(Double.class)
                .metavar("THROUGHPUT")
                .help("throttle maximum message throughput to *approximately* THROUGHPUT messages/sec. Set this to -1 to disable throttling.");

        parser.addArgument("--producer-props")
                 .nargs("+")
                 .required(false)
                 .metavar("PROP-NAME=PROP-VALUE")
                 .type(String.class)
                 .dest("producerConfig")
                 .help("kafka producer related configuration properties like bootstrap.servers,client.id etc. " +
                         "These configs take precedence over those passed via --producer.config.");

        parser.addArgument("--producer.config")
                .action(store())
                .required(false)
                .type(String.class)
                .metavar("CONFIG-FILE")
                .dest("producerConfigFile")
                .help("producer config properties file.");

        parser.addArgument("--print-metrics")
                .action(storeTrue())
                .type(Boolean.class)
                .metavar("PRINT-METRICS")
                .dest("printMetrics")
                .help("print out metrics at the end of the test.");

        parser.addArgument("--transactional-id")
               .action(store())
               .required(false)
               .type(String.class)
               .metavar("TRANSACTIONAL-ID")
               .dest("transactionalId")
               .help("The transactional id to use. This config takes precedence over the transactional.id " +
                       "specified via --producer.config or --producer-props. Note that if the transactional id " +
                       "is not specified while --transaction-duration-ms is provided, the default value for the " +
                       "transactional id will be performance-producer- followed by a random uuid.");

        parser.addArgument("--transaction-duration-ms")
               .action(store())
               .required(false)
               .type(Long.class)
               .metavar("TRANSACTION-DURATION")
               .dest("transactionDurationMs")
               .help("The max age of each transaction. The commitTransaction will be called after this time has elapsed. " +
                       "The value should be greater than 0. If the transactional id is specified via --producer-props, " +
                       "--producer.config, or --transactional-id but --transaction-duration-ms is not specified, " +
                       "the default value will be 3000.");

        return parser;
    }
@Test
    public void testEnableTransactionByProducerProps() throws IOException, ArgumentParserException {
        ArgumentParser parser = ProducerPerformance.argParser();
        String[] args = new String[]{
            "--topic", "Hello-Kafka",
            "--num-records", "5",
            "--throughput", "100",
            "--record-size", "100",
            "--producer-props", "bootstrap.servers=localhost:9000", "transactional.id=foobar"};
        ProducerPerformance.ConfigPostProcessor configs = new ProducerPerformance.ConfigPostProcessor(parser, args);
        assertTrue(configs.transactionsEnabled);
        assertEquals(ProducerPerformance.DEFAULT_TRANSACTION_DURATION_MS, configs.transactionDurationMs);
        assertEquals("foobar", configs.producerProps.get(ProducerConfig.TRANSACTIONAL_ID_CONFIG));
    }