Repository: apache/kafka
File: server/src/test/java/org/apache/kafka/server/BrokerRegistrationRequestTest.java (Java)

Code before:

    @ClusterTest(types = {Type.KRAFT}, controllers = 1, metadataVersion = MetadataVersion.IBP_3_3_IV3)
    public void shouldRegisterWhenSupportedRangeAndNotMigrating(ClusterInstance clusterInstance) throws Exception {
        var clusterId = clusterInstance.clusterId();
        var channelManager = brokerToControllerChannelManager(clusterInstance);
        try {
            channelManager.start();
            Assertions.assertEquals(
                Errors.NONE,
                registerBroker(channelManager, clusterId, 100, null, 
                    new FeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel(), MetadataVersion.IBP_3_4_IV0.featureLevel()))
            );
        } finally {
            channelManager.shutdown();
        }
    }

    public record FeatureLevel(short min, short max) { }

    private record TestControllerNodeProvider(SocketServer controllerSocketServer, ClusterInstance clusterInstance)
            implements ControllerNodeProvider {

        public Optional<Node> node() {
            return Optional.of(new Node(
                    controllerSocketServer.config().nodeId(),
                    "127.0.0.1",
                    controllerSocketServer.boundPort(clusterInstance.controllerListenerName())
            ));
        }

        public ListenerName listenerName() {
            return clusterInstance.controllerListenerName();
        }

        public SecurityProtocol securityProtocol() {
            return SecurityProtocol.PLAINTEXT;
        }

        public String saslMechanism() {
            return "";
        }

        @Override
        public ControllerInformation getControllerInfo() {
            return ControllerInformation.apply(
                    Option.apply(node().orElse(null)),
                    listenerName(),
                    securityProtocol(),
                    saslMechanism()
            );
        }

Review comment (suggestion):
```java
        public Optional<Node> node() {
            return Optional.of(new Node(
                    clusterInstance.controllers().keySet().iterator().next(),
                    "127.0.0.1",
                    clusterInstance.controllerBoundPorts().get(0)
            ));
        }
```

Code after:
                .setName("EXTERNAL")
                .setHost("example.com")
                .setPort(8082)
                .setSecurityProtocol(SecurityProtocol.PLAINTEXT.id);

        var req = new BrokerRegistrationRequestData()
                .setBrokerId(100)
                .setLogDirs(List.of(Uuid.randomUuid()))
                .setClusterId(clusterInstance.clusterId())
                .setIncarnationId(Uuid.randomUuid())
                .setIsMigratingZkBroker(zkEpoch != null)
                .setFeatures(features)
                .setListeners(new BrokerRegistrationRequestData.ListenerCollection(List.of(listener).iterator()));

        BrokerRegistrationResponse resp = this.sendAndReceive(
                channelManager,
                new BrokerRegistrationRequest.Builder(req)
        );
        return Errors.forCode(resp.data().errorCode());
    }

    record TestControllerNodeProvider(ClusterInstance clusterInstance)
            implements ControllerNodeProvider {

        public Optional<Node> node() {
            return Optional.of(new Node(
                    clusterInstance.controllers().keySet().iterator().next(),
                    "127.0.0.1",
                    clusterInstance.controllerBoundPorts().get(0)
            ));
        }
        
        public ListenerName listenerName() {
            return clusterInstance.controllerListenerName();
        }

        public SecurityProtocol securityProtocol() {
            return SecurityProtocol.PLAINTEXT;
        }

        public String saslMechanism() {
            return "";
        }

        @Override
        public ControllerInformation getControllerInfo() {
            return ControllerInformation.apply(
                    Option.apply(node().orElse(null)),
                    listenerName(),
                    securityProtocol(),
                    saslMechanism()