private KafkaConfig createNodeConfig(TestKitNode node) {
            BrokerNode brokerNode = nodes.brokerNodes().get(node.id());
            ControllerNode controllerNode = nodes.controllerNodes().get(node.id());

            Map<String, Object> props = new HashMap<>(configProps);
            props.put(KRaftConfigs.SERVER_MAX_STARTUP_TIME_MS_CONFIG,
                    Long.toString(TimeUnit.MINUTES.toMillis(10)));
            props.put(KRaftConfigs.PROCESS_ROLES_CONFIG, roles(node.id()));
            props.put(KRaftConfigs.NODE_ID_CONFIG,
                    Integer.toString(node.id()));
            // In combined mode, always prefer the metadata log directory of the controller node.
            if (controllerNode != null) {
                props.put(KRaftConfigs.METADATA_LOG_DIR_CONFIG,
                        controllerNode.metadataDirectory());
            } else {
                props.put(KRaftConfigs.METADATA_LOG_DIR_CONFIG,
                        node.metadataDirectory());
            }
            if (brokerNode != null) {
                // Set the log.dirs according to the broker node setting (if there is a broker node)
                props.put(LOG_DIRS_CONFIG,
                        String.join(",", brokerNode.logDataDirectories()));
            } else {
                // Set log.dirs equal to the metadata directory if there is just a controller.
                props.put(LOG_DIRS_CONFIG,
                    controllerNode.metadataDirectory());
            }

            // We allow configuring the listeners and related properties via Builder::setConfigProp,
            // and they shouldn't be overridden here
            props.putIfAbsent(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG,
                "EXTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT");
            props.putIfAbsent(SocketServerConfigs.LISTENERS_CONFIG, listeners(node.id()));
            props.putIfAbsent(INTER_BROKER_LISTENER_NAME_CONFIG,
                nodes.interBrokerListenerName().value());
            props.putIfAbsent(KRaftConfigs.CONTROLLER_LISTENER_NAMES_CONFIG, "CONTROLLER");

            // Note: we can't accurately set controller.quorum.voters yet, since we don't
            // yet know what ports each controller will pick.  Set it to a dummy string
            // for now as a placeholder.
            String uninitializedQuorumVotersString = nodes.controllerNodes().keySet().stream().
                    map(n -> String.format("%d@0.0.0.0:0", n)).
                    collect(Collectors.joining(","));
            props.put(QuorumConfig.QUORUM_VOTERS_CONFIG, uninitializedQuorumVotersString);

            // reduce log cleaner offset map memory usage
            props.put(CleanerConfig.LOG_CLEANER_DEDUPE_BUFFER_SIZE_PROP, "2097152");

            // Add associated broker node property overrides
            if (brokerNode != null) {
                props.putAll(brokerNode.propertyOverrides());
            }
            // Add associated controller node property overrides
            if (controllerNode != null) {
                props.putAll(controllerNode.propertyOverrides());
            }
            props.putIfAbsent(ServerConfigs.UNSTABLE_FEATURE_VERSIONS_ENABLE_CONFIG, "true");
            props.putIfAbsent(ServerConfigs.UNSTABLE_API_VERSIONS_ENABLE_CONFIG, "true");
            return new KafkaConfig(props, false);
        }
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