void testAdminClient() throws Throwable {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, testConfig.bootstrapServer);
        try (final Admin client = Admin.create(adminProps)) {
            while (true) {
                Collection<Node> nodes = client.describeCluster().nodes().get();
                if (nodes.size() == testConfig.numClusterNodes) {
                    break;
                } else if (nodes.size() > testConfig.numClusterNodes) {
                    throw new KafkaException("Expected to see " + testConfig.numClusterNodes +
                        " nodes, but saw " + nodes.size());
                }
                Thread.sleep(1);
                log.info("Saw only {} cluster nodes.  Waiting to see {}.",
                    nodes.size(), testConfig.numClusterNodes);
            }

            testDescribeConfigsMethod(client);

            tryFeature("createTopics", testConfig.createTopicsSupported,
                () -> {
                    try {
                        client.createTopics(Collections.singleton(
                            new NewTopic("newtopic", 1, (short) 1))).all().get();
                    } catch (ExecutionException e) {
                        throw e.getCause();
                    }
                },
                () ->  createTopicsResultTest(client, Collections.singleton("newtopic"))
            );

            while (true) {
                Collection<TopicListing> listings = client.listTopics().listings().get();
                if (!testConfig.createTopicsSupported)
                    break;

                if (topicExists(listings, "newtopic"))
                    break;

                Thread.sleep(1);
                log.info("Did not see newtopic.  Retrying listTopics...");
            }

            tryFeature("describeAclsSupported", testConfig.describeAclsSupported,
                () -> {
                    try {
                        client.describeAcls(AclBindingFilter.ANY).values().get();
                    } catch (ExecutionException e) {
                        if (e.getCause() instanceof SecurityDisabledException)
                            return;
                        throw e.getCause();
                    }
                });
        }
    }
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