private static ArgumentParser argParser() {
        ArgumentParser parser = ArgumentParsers
                .newArgumentParser("transactional-message-copier")
                .description("This tool copies messages transactionally from an input partition to an output topic, " +
                        "committing the consumed offsets along with the output messages");

        parser.addArgument("--input-topic")
                .action(store())
                .required(true)
                .type(String.class)
                .metavar("INPUT-TOPIC")
                .dest("inputTopic")
                .help("Consume messages from this topic");

        parser.addArgument("--input-partition")
                .action(store())
                .required(true)
                .type(Integer.class)
                .metavar("INPUT-PARTITION")
                .dest("inputPartition")
                .help("Consume messages from this partition of the input topic.");

        parser.addArgument("--output-topic")
                .action(store())
                .required(true)
                .type(String.class)
                .metavar("OUTPUT-TOPIC")
                .dest("outputTopic")
                .help("Produce messages to this topic");

        parser.addArgument("--broker-list")
                .action(store())
                .required(true)
                .type(String.class)
                .metavar("HOST1:PORT1[,HOST2:PORT2[...]]")
                .dest("brokerList")
                .help("Comma-separated list of Kafka brokers in the form HOST1:PORT1,HOST2:PORT2,...");

        parser.addArgument("--max-messages")
                .action(store())
                .required(false)
                .setDefault(-1)
                .type(Integer.class)
                .metavar("MAX-MESSAGES")
                .dest("maxMessages")
                .help("Process these many messages upto the end offset at the time this program was launched. If set to -1 " +
                        "we will just read to the end offset of the input partition (as of the time the program was launched).");

        parser.addArgument("--consumer-group")
                .action(store())
                .required(false)
                .setDefault(-1)
                .type(String.class)
                .metavar("CONSUMER-GROUP")
                .dest("consumerGroup")
                .help("The consumer group id to use for storing the consumer offsets.");

        parser.addArgument("--transaction-size")
                .action(store())
                .required(false)
                .setDefault(200)
                .type(Integer.class)
                .metavar("TRANSACTION-SIZE")
                .dest("messagesPerTransaction")
                .help("The number of messages to put in each transaction. Default is 200.");

        parser.addArgument("--transaction-timeout")
                .action(store())
                .required(false)
                .setDefault(60000)
                .type(Integer.class)
                .metavar("TRANSACTION-TIMEOUT")
                .dest("transactionTimeout")
                .help("The transaction timeout in milliseconds. Default is 60000(1 minute).");

        parser.addArgument("--transactional-id")
                .action(store())
                .required(true)
                .type(String.class)
                .metavar("TRANSACTIONAL-ID")
                .dest("transactionalId")
                .help("The transactionalId to assign to the producer");

        parser.addArgument("--producer-block-timeout-ms")
                .action(store())
                .required(false)
                .setDefault(60000)
                .type(Integer.class)
                .metavar("PRODUCER-BLOCK-TIMEOUT-MS")
                .dest("producerBlockTimeoutMs")
                .help("The maximum time in milliseconds the producer will block for during a send request.");

        parser.addArgument("--enable-random-aborts")
                .action(storeTrue())
                .type(Boolean.class)
                .metavar("ENABLE-RANDOM-ABORTS")
                .dest("enableRandomAborts")
                .help("Whether or not to enable random transaction aborts (for system testing)");

        parser.addArgument("--group-mode")
                .action(storeTrue())
                .type(Boolean.class)
                .metavar("GROUP-MODE")
                .dest("groupMode")
                .help("Whether to let consumer subscribe to the input topic or do manual assign. If we do" +
                          " subscription based consumption, the input partition shall be ignored");

        parser.addArgument("--use-group-metadata")
                .action(storeTrue())
                .type(Boolean.class)
                .metavar("USE-GROUP-METADATA")
                .dest("useGroupMetadata")
                .help("Whether to use the new transactional commit API with group metadata");

        parser.addArgument("--consumer-default-api-timeout-ms")
                .action(store())
                .required(false)
                .setDefault(60000)
                .type(Integer.class)
                .metavar("CONSUMER-DEFAULT-API-TIMEOUT-MS")
                .dest("consumerDefaultApiTimeoutMs")
                .help("The default API timeout in milliseconds for the consumer.");

        return parser;
    }
Get the command-line argument parser.
tools/src/main/java/org/apache/kafka/tools/TransactionalMessageCopier.java