Repository: apache/kafka
File: core/src/test/scala/unit/kafka/server/BrokerRegistrationRequestTest.scala (Scala)

Code before:
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.test.ClusterInstance
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.{Node, Uuid}
import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, Feature, MetadataVersion, MetadataVersionTestUtils, NodeToControllerChannelManager}
import org.junit.jupiter.api.Assertions.assertEquals

import java.util
import java.util.Collections
import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}

/**
 * This test simulates a broker registering with the KRaft quorum under different configurations.
 */
class BrokerRegistrationRequestTest {

  def brokerToControllerChannelManager(clusterInstance: ClusterInstance): NodeToControllerChannelManager = {
    def anyControllerSocketServer(clusterInstance: ClusterInstance): SocketServer = {
      clusterInstance.controllers().values().stream().map(_.socketServer).findFirst().get()
    }
    new NodeToControllerChannelManagerImpl(
      new ControllerNodeProvider() {
        def node: Option[Node] = Some(new Node(
          anyControllerSocketServer(clusterInstance).config.nodeId,
          "127.0.0.1",
          anyControllerSocketServer(clusterInstance).boundPort(clusterInstance.controllerListenerName()),
        ))

        def listenerName: ListenerName = clusterInstance.controllerListenerName()

        val securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT

        val saslMechanism: String = ""

        override def getControllerInfo(): ControllerInformation =
          ControllerInformation(node, listenerName, securityProtocol, saslMechanism)
      },
      Time.SYSTEM,
      new Metrics(),
      anyControllerSocketServer(clusterInstance).config,
      "heartbeat",
      "test-heartbeat-",
      10000
    )
  }

  def sendAndReceive[T <: AbstractRequest, R <: AbstractResponse](
    channelManager: NodeToControllerChannelManager,
    reqBuilder: AbstractRequest.Builder[T],
    timeoutMs: Int
  ): R = {

Review comment (bug):
this looks like a bug. It could create a nonexistent node, right? the port and broker id could be unmatched

Code after:
import org.apache.kafka.common.test.ClusterInstance
import org.apache.kafka.common.utils.Time
import org.apache.kafka.common.{Node, Uuid}
import org.apache.kafka.server.common.{ControllerRequestCompletionHandler, Feature, MetadataVersion, MetadataVersionTestUtils, NodeToControllerChannelManager}
import org.junit.jupiter.api.Assertions.assertEquals

import java.util
import java.util.Collections
import java.util.concurrent.{CompletableFuture, TimeUnit, TimeoutException}

/**
 * This test simulates a broker registering with the KRaft quorum under different configurations.
 */
class BrokerRegistrationRequestTest {

  def brokerToControllerChannelManager(clusterInstance: ClusterInstance): NodeToControllerChannelManager = {
    val controllerSocketServer = clusterInstance.controllers().values().stream().map(_.socketServer).findFirst().get()
    new NodeToControllerChannelManagerImpl(
      new ControllerNodeProvider() {
        def node: Option[Node] = Some(new Node(
          controllerSocketServer.config.nodeId,
          "127.0.0.1",
          controllerSocketServer.boundPort(clusterInstance.controllerListenerName()),
        ))

        def listenerName: ListenerName = clusterInstance.controllerListenerName()

        val securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT

        val saslMechanism: String = ""

        override def getControllerInfo(): ControllerInformation =
          ControllerInformation(node, listenerName, securityProtocol, saslMechanism)
      },
      Time.SYSTEM,
      new Metrics(),
      controllerSocketServer.config,
      "heartbeat",
      "test-heartbeat-",
      10000
    )
  }

  def sendAndReceive[T <: AbstractRequest, R <: AbstractResponse](
    channelManager: NodeToControllerChannelManager,
    reqBuilder: AbstractRequest.Builder[T],
    timeoutMs: Int
  ): R = {
    val responseFuture = new CompletableFuture[R]()
    channelManager.sendRequest(reqBuilder, new ControllerRequestCompletionHandler() {
      override def onTimeout(): Unit = responseFuture.completeExceptionally(new TimeoutException())