Repository: apache/kafka
File: test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/KafkaClusterTestKit.java (Java)

Code before:
            Formatter formatter = new Formatter();
            formatter.setNodeId(nodeId);
            formatter.setClusterId(ensemble.clusterId().get());
            if (writeMetadataDirectory) {
                formatter.setDirectories(ensemble.logDirProps().keySet());
            } else {
                formatter.setDirectories(ensemble.logDirProps().keySet().stream().
                    filter(d -> !ensemble.metadataLogDir().get().equals(d)).
                    collect(Collectors.toSet()));
            }
            if (formatter.directories().isEmpty()) {
                return;
            }
            formatter.setReleaseVersion(nodes.bootstrapMetadata().metadataVersion());
            formatter.setUnstableFeatureVersionsEnabled(true);
            formatter.setIgnoreFormatted(false);
            formatter.setControllerListenerName(controllerListenerName);
            if (writeMetadataDirectory) {
                formatter.setMetadataLogDirectory(ensemble.metadataLogDir().get());
            } else {
                formatter.setMetadataLogDirectory(Optional.empty());
            }
            StringBuilder dynamicVotersBuilder = new StringBuilder();
            String prefix = "";
            if (standalone) {
                if (nodeId == TestKitDefaults.CONTROLLER_ID_OFFSET) {
                    final var controllerNode = nodes.controllerNodes().get(nodeId);
                    dynamicVotersBuilder.append(
                        String.format(
                            "%d@localhost:%d:%s",
                            controllerNode.id(),
                            socketFactoryManager.
                                getOrCreatePortForListener(controllerNode.id(), controllerListenerName),
                            controllerNode.metadataDirectoryId()
                        )
                    );
                    formatter.setInitialControllers(DynamicVoters.parse(dynamicVotersBuilder.toString()));
                }
                // when the nodeId != TestKitDefaults.CONTROLLER_ID_OFFSET, the node is formatting with
                // the --no-initial-controllers flag
                formatter.setHasDynamicQuorum(true);
            } else if (initialVoterSet.isPresent()) {
                for (final var controllerNode : initialVoterSet.get().entrySet()) {
                    final var voterId = controllerNode.getKey();
                    final var voterDirectoryId = controllerNode.getValue();
                    dynamicVotersBuilder.append(prefix);
                    prefix = ",";
                    dynamicVotersBuilder.append(
                        String.format(
                            "%d@localhost:%d:%s",
                            voterId,

Review comment (suggestion):
The true port of the standalone controller is `TestKitDefaults.BROKER_ID_OFFSET + TestKitDefaults.CONTROLLER_ID_OFFSET`. We will likely not change the value of `BROKER_ID_OFFSET` in the future, but should we still use `BROKER_ID_OFFSET + CONTROLLER_ID_OFFSET` instead?

https://github.com/apache/kafka/blob/trunk/test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestKitNodes.java#L175

Code after:
            Formatter formatter = new Formatter();
            formatter.setNodeId(nodeId);
            formatter.setClusterId(ensemble.clusterId().get());
            if (writeMetadataDirectory) {
                formatter.setDirectories(ensemble.logDirProps().keySet());
            } else {
                formatter.setDirectories(ensemble.logDirProps().keySet().stream().
                    filter(d -> !ensemble.metadataLogDir().get().equals(d)).
                    collect(Collectors.toSet()));
            }
            if (formatter.directories().isEmpty()) {
                return;
            }
            formatter.setReleaseVersion(nodes.bootstrapMetadata().metadataVersion());
            formatter.setUnstableFeatureVersionsEnabled(true);
            formatter.setIgnoreFormatted(false);
            formatter.setControllerListenerName(controllerListenerName);
            if (writeMetadataDirectory) {
                formatter.setMetadataLogDirectory(ensemble.metadataLogDir().get());
            } else {
                formatter.setMetadataLogDirectory(Optional.empty());
            }
            StringBuilder dynamicVotersBuilder = new StringBuilder();
            String prefix = "";
            if (standalone) {
                if (nodeId == TestKitDefaults.BROKER_ID_OFFSET + TestKitDefaults.CONTROLLER_ID_OFFSET) {
                    final var controllerNode = nodes.controllerNodes().get(nodeId);
                    dynamicVotersBuilder.append(
                        String.format(
                            "%d@localhost:%d:%s",
                            controllerNode.id(),
                            socketFactoryManager.
                                getOrCreatePortForListener(controllerNode.id(), controllerListenerName),
                            controllerNode.metadataDirectoryId()
                        )
                    );
                    formatter.setInitialControllers(DynamicVoters.parse(dynamicVotersBuilder.toString()));
                }
                // when the nodeId != TestKitDefaults.CONTROLLER_ID_OFFSET, the node is formatting with
                // the --no-initial-controllers flag
                formatter.setHasDynamicQuorum(true);
            } else if (initialVoterSet.isPresent()) {
                for (final var controllerNode : initialVoterSet.get().entrySet()) {
                    final var voterId = controllerNode.getKey();
                    final var voterDirectoryId = controllerNode.getValue();
                    dynamicVotersBuilder.append(prefix);
                    prefix = ",";
                    dynamicVotersBuilder.append(
                        String.format(
                            "%d@localhost:%d:%s",
                            voterId,