public KafkaClusterTestKit build() throws Exception {
            Map<Integer, ControllerServer> controllers = new HashMap<>();
            Map<Integer, BrokerServer> brokers = new HashMap<>();
            Map<Integer, SharedServer> jointServers = new HashMap<>();
            /*
              Number of threads = Total number of brokers + Total number of controllers + Total number of Raft Managers
                                = Total number of brokers + Total number of controllers * 2
                                  (Raft Manager per broker/controller)
             */
            int numOfExecutorThreads = (nodes.brokerNodes().size() + nodes.controllerNodes().size()) * 2;
            ExecutorService executorService = null;
            ControllerQuorumVotersFutureManager connectFutureManager =
                new ControllerQuorumVotersFutureManager(nodes.controllerNodes().size());
            File baseDirectory = null;

            try {
                baseDirectory = new File(nodes.baseDirectory());
                executorService = Executors.newFixedThreadPool(numOfExecutorThreads,
                    ThreadUtils.createThreadFactory("kafka-cluster-test-kit-executor-%d", false));
                for (ControllerNode node : nodes.controllerNodes().values()) {
                    setupNodeDirectories(baseDirectory, node.metadataDirectory(), Collections.emptyList());
                    SharedServer sharedServer = new SharedServer(
                        createNodeConfig(node),
                        node.initialMetaPropertiesEnsemble(),
                        Time.SYSTEM,
                        new Metrics(),
                        connectFutureManager.future,
                        Collections.emptyList(),
                        faultHandlerFactory
                    );
                    ControllerServer controller = null;
                    try {
                        controller = new ControllerServer(
                                sharedServer,
                                KafkaRaftServer.configSchema(),
                                nodes.bootstrapMetadata());
                    } catch (Throwable e) {
                        log.error("Error creating controller {}", node.id(), e);
                        Utils.swallow(log, Level.WARN, "sharedServer.stopForController error", sharedServer::stopForController);
                        throw e;
                    }
                    controllers.put(node.id(), controller);
                    controller.socketServerFirstBoundPortFuture().whenComplete((port, e) -> {
                        if (e != null) {
                            connectFutureManager.fail(e);
                        } else {
                            connectFutureManager.registerPort(node.id(), port);
                        }
                    });
                    jointServers.put(node.id(), sharedServer);
                }
                for (BrokerNode node : nodes.brokerNodes().values()) {
                    SharedServer sharedServer = jointServers.computeIfAbsent(
                        node.id(),
                        id -> new SharedServer(
                            createNodeConfig(node),
                            node.initialMetaPropertiesEnsemble(),
                            Time.SYSTEM,
                            new Metrics(),
                            connectFutureManager.future,
                            Collections.emptyList(),
                            faultHandlerFactory
                        )
                    );
                    BrokerServer broker = null;
                    try {
                        broker = new BrokerServer(sharedServer);
                    } catch (Throwable e) {
                        log.error("Error creating broker {}", node.id(), e);
                        Utils.swallow(log, Level.WARN, "sharedServer.stopForBroker error", sharedServer::stopForBroker);
                        throw e;
                    }
                    brokers.put(node.id(), broker);
                }
            } catch (Exception e) {
                if (executorService != null) {
                    ThreadUtils.shutdownExecutorServiceQuietly(executorService, 5, TimeUnit.MINUTES);
                }
                for (BrokerServer brokerServer : brokers.values()) {
                    brokerServer.shutdown();
                }
                for (ControllerServer controller : controllers.values()) {
                    controller.shutdown();
                }
                connectFutureManager.close();
                if (baseDirectory != null) {
                    Utils.delete(baseDirectory);
                }
                throw e;
            }
            return new KafkaClusterTestKit(executorService,
                    nodes,
                    controllers,
                    brokers,
                    connectFutureManager,
                    baseDirectory,
                    faultHandlerFactory);
        }
This class manages a future which is completed with the proper value for
controller.quorum.voters once the randomly assigned ports for all the controllers are
known.
core/src/test/java/kafka/testkit/KafkaClusterTestKit.java