Repository: apache/kafka
File: core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala (Scala)

Code before:
        s"but only ${runningBrokerServers.size} broker(s) are.")
    runningBrokerServers
  }

  private def sendDescribeClusterRequestToBoundPortUntilBrokersPropagated(destination: BrokerServer,
                                                                          listenerName: ListenerName,
                                                                          expectedBrokerCount: Int,
                                                                          waitTime: FiniteDuration): DescribeClusterResponse = {
    val (describeClusterResponse, metadataUpToDate) = TestUtils.computeUntilTrue(
      compute = sendDescribeClusterRequestToBoundPort(destination.socketServer, listenerName),
      waitTime = waitTime.toMillis
    ) {
      response => response.nodes.size == expectedBrokerCount
    }

    assertTrue(metadataUpToDate,
      s"After ${waitTime.toMillis} ms Broker is only aware of ${describeClusterResponse.nodes.size} brokers, " +
        s"but $expectedBrokerCount are expected.")

    describeClusterResponse
  }

  private def sendDescribeClusterRequestToBoundPort(destination: SocketServer,
                                                    listenerName: ListenerName): DescribeClusterResponse = {

      val port = destination.boundPort(listenerName)

      connectAndReceive[DescribeClusterResponse](
            request = new DescribeClusterRequest.Builder(new DescribeClusterRequestData()).build(),
            port
          )
  }


  @Test
  def testCreateClusterAndPerformReassignment(): Unit = {
    val cluster = new KafkaClusterTestKit.Builder(
      new TestKitNodes.Builder().
        setNumBrokerNodes(4).
        setNumControllerNodes(3).build()).build()
    try {
      cluster.format()
      cluster.startup()
      cluster.waitForReadyBrokers()
      val admin = Admin.create(cluster.clientProperties())
      try {
        // Create the topic.
        val assignments = new util.HashMap[Integer, util.List[Integer]]
        assignments.put(0, util.List.of(0, 1, 2))
        assignments.put(1, util.List.of(1, 2, 3))
        assignments.put(2, util.List.of(2, 3, 0))

Review comment (suggestion):
```scala
      connectAndReceive[DescribeClusterResponse](new DescribeClusterRequest.Builder(new DescribeClusterRequestData()).build(),
        destination.boundPort(listenerName))
```

Code after:
        s"but only ${runningBrokerServers.size} broker(s) are.")
    runningBrokerServers
  }

  private def sendDescribeClusterRequestToBoundPortUntilBrokersPropagated(destination: BrokerServer,
                                                                          listenerName: ListenerName,
                                                                          expectedBrokerCount: Int,
                                                                          waitTime: FiniteDuration): DescribeClusterResponse = {
    val (describeClusterResponse, metadataUpToDate) = TestUtils.computeUntilTrue(
      compute = sendDescribeClusterRequestToBoundPort(destination.socketServer, listenerName),
      waitTime = waitTime.toMillis
    ) {
      response => response.nodes.size == expectedBrokerCount
    }

    assertTrue(metadataUpToDate,
      s"After ${waitTime.toMillis} ms Broker is only aware of ${describeClusterResponse.nodes.size} brokers, " +
        s"but $expectedBrokerCount are expected.")

    describeClusterResponse
  }

  private def sendDescribeClusterRequestToBoundPort(destination: SocketServer,
                                                    listenerName: ListenerName): DescribeClusterResponse = {
    connectAndReceive[DescribeClusterResponse](new DescribeClusterRequest.Builder(new DescribeClusterRequestData()).build(),
      destination.boundPort(listenerName))
  }

  @Test
  def testCreateClusterAndPerformReassignment(): Unit = {
    val cluster = new KafkaClusterTestKit.Builder(
      new TestKitNodes.Builder().
        setNumBrokerNodes(4).
        setNumControllerNodes(3).build()).build()
    try {
      cluster.format()
      cluster.startup()
      cluster.waitForReadyBrokers()
      val admin = Admin.create(cluster.clientProperties())
      try {
        // Create the topic.
        val assignments = new util.HashMap[Integer, util.List[Integer]]
        assignments.put(0, util.List.of(0, 1, 2))
        assignments.put(1, util.List.of(1, 2, 3))
        assignments.put(2, util.List.of(2, 3, 0))
        assignments.put(3, util.List.of(3, 2, 1))
        val createTopicResult = admin.createTopics(util.List.of(
          new NewTopic("foo", assignments)))
        createTopicResult.all().get()
        waitForTopicListing(admin, Seq("foo"), Seq())