public static void main(String[] args) throws Exception {
        ArgumentParser parser = ArgumentParsers
            .newArgumentParser("client-compatibility-test")
            .description("This tool is used to verify client compatibility guarantees.");
        parser.addArgument("--topic")
            .action(store())
            .required(true)
            .type(String.class)
            .dest("topic")
            .metavar("TOPIC")
            .help("the compatibility test will produce messages to this topic");
        parser.addArgument("--bootstrap-server")
            .action(store())
            .required(true)
            .type(String.class)
            .dest("bootstrapServer")
            .metavar("BOOTSTRAP_SERVER")
            .help("The server(s) to use for bootstrapping");
        parser.addArgument("--offsets-for-times-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("offsetsForTimesSupported")
            .metavar("OFFSETS_FOR_TIMES_SUPPORTED")
            .help("True if KafkaConsumer#offsetsForTimes is supported by the current broker version");
        parser.addArgument("--cluster-id-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("clusterIdSupported")
            .metavar("CLUSTER_ID_SUPPORTED")
            .help("True if cluster IDs are supported.  False if cluster ID always appears as null.");
        parser.addArgument("--expect-record-too-large-exception")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("expectRecordTooLargeException")
            .metavar("EXPECT_RECORD_TOO_LARGE_EXCEPTION")
            .help("True if we should expect a RecordTooLargeException when trying to read from a topic " +
                  "that contains a message that is bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG +
                  ".  This is pre-KIP-74 behavior.");
        parser.addArgument("--num-cluster-nodes")
            .action(store())
            .required(true)
            .type(Integer.class)
            .dest("numClusterNodes")
            .metavar("NUM_CLUSTER_NODES")
            .help("The number of cluster nodes we should expect to see from the AdminClient.");
        parser.addArgument("--create-topics-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("createTopicsSupported")
            .metavar("CREATE_TOPICS_SUPPORTED")
            .help("Whether we should be able to create topics via the AdminClient.");
        parser.addArgument("--describe-acls-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("describeAclsSupported")
            .metavar("DESCRIBE_ACLS_SUPPORTED")
            .help("Whether describeAcls is supported in the AdminClient.");
        parser.addArgument("--describe-configs-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("describeConfigsSupported")
            .metavar("DESCRIBE_CONFIGS_SUPPORTED")
            .help("Whether describeConfigs is supported in the AdminClient.");
        parser.addArgument("--idempotent-producer-supported")
            .action(store())
            .required(true)
            .type(Boolean.class)
            .dest("idempotentProducerSupported")
            .metavar("IDEMPOTENT_PRODUCER_SUPPORTED")
            .help("Whether the producer supports idempotency.");

        Namespace res = null;
        try {
            res = parser.parseArgs(args);
        } catch (ArgumentParserException e) {
            if (args.length == 0) {
                parser.printHelp();
                Exit.exit(0);
            } else {
                parser.handleError(e);
                Exit.exit(1);
            }
        }
        TestConfig testConfig = new TestConfig(res);
        ClientCompatibilityTest test = new ClientCompatibilityTest(testConfig);
        try {
            test.run();
        } catch (Throwable t) {
            System.out.printf("FAILED: Caught exception %s%n%n", t.getMessage());
            t.printStackTrace();
            Exit.exit(1);
        }
        System.out.println("SUCCESS.");
        Exit.exit(0);
    }
ClientCompatibilityTest is invoked by the ducktape test client_compatibility_features_test.py to validate
client behavior when various broker versions are in use.  It runs various client operations and tests whether they
are supported or not.
tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java