public void format() throws Exception {
            if (formated.compareAndSet(false, true)) {
                List<ApiMessageAndVersion> records = new ArrayList<>();
                records.add(
                    new ApiMessageAndVersion(new FeatureLevelRecord().
                        setName(MetadataVersion.FEATURE_NAME).
                        setFeatureLevel(clusterConfig.metadataVersion().featureLevel()), (short) 0));

                clusterConfig.features().forEach((feature, version) -> {
                    records.add(
                        new ApiMessageAndVersion(new FeatureLevelRecord().
                            setName(feature.featureName()).
                            setFeatureLevel(version), (short) 0));
                });

                TestKitNodes nodes = new TestKitNodes.Builder()
                        .setBootstrapMetadata(BootstrapMetadata.fromRecords(records, "testkit"))
                        .setCombined(isCombined)
                        .setNumBrokerNodes(clusterConfig.numBrokers())
                        .setNumDisksPerBroker(clusterConfig.numDisksPerBroker())
                        .setPerServerProperties(clusterConfig.perServerOverrideProperties())
                        .setNumControllerNodes(clusterConfig.numControllers()).build();
                KafkaClusterTestKit.Builder builder = new KafkaClusterTestKit.Builder(nodes);
                if (Boolean.parseBoolean(clusterConfig.serverProperties()
                        .getOrDefault("zookeeper.metadata.migration.enable", "false"))) {
                    this.embeddedZookeeper = new EmbeddedZookeeper();
                    builder.setConfigProp("zookeeper.connect", String.format("localhost:%d", embeddedZookeeper.port()));
                }
                // Copy properties into the TestKit builder
                clusterConfig.serverProperties().forEach(builder::setConfigProp);
                // KAFKA-12512 need to pass security protocol and listener name here
                this.clusterTestKit = builder.build();
                this.clusterTestKit.format();
            }
        }
Wraps a {@link KafkaClusterTestKit} inside lifecycle methods for a test invocation. Each instance of this
class is provided with a configuration for the cluster.

This context also provides parameter resolvers for:

<ul>
    <li>ClusterConfig (the same instance passed to the constructor)</li>
    <li>ClusterInstance (includes methods to expose underlying SocketServer-s)</li>
</ul>
core/src/test/java/kafka/test/junit/RaftClusterInvocationContext.java