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

Code before:
          .setGroupId(groupIds(i))
          .setErrorCode(expectedErrors(i).code)
      }.asJava.iterator))

    val deleteGroupsResponse = connectAndReceive[DeleteGroupsResponse](deleteGroupsRequest)
    assertEquals(expectedResponseData.results.asScala.toSet, deleteGroupsResponse.data.results.asScala.toSet)
  }

  protected def connectAny(): Socket = {
    val socket: Socket = IntegrationTestUtils.connect(cluster.boundPorts().get(0))
    openSockets += socket
    socket
  }

  protected def connect(destination: Int): Socket = {
    val socket = IntegrationTestUtils.connect(brokerSocketServer(destination).boundPort(cluster.clientListener()))
    openSockets += socket
    socket
  }

  protected def connectAndReceive[T <: AbstractResponse](
    request: AbstractRequest
  ): T = {
    val destination = cluster.anyBrokerSocketServer()
    val listenerName = cluster.clientListener()
    IntegrationTestUtils.connectAndReceive[T](request, destination.boundPort(listenerName))
  }

  protected def connectAndReceive[T <: AbstractResponse](
    request: AbstractRequest,
    destination: Int
  ): T = {
    val socketServer = brokerSocketServer(destination)
    val listenerName = cluster.clientListener()
    IntegrationTestUtils.connectAndReceive[T](request, socketServer.boundPort(listenerName))
  }

  private def brokerSocketServer(brokerId: Int): SocketServer = {
    getBrokers.find { broker =>
      broker.config.brokerId == brokerId
    }.map(_.socketServer).getOrElse(throw new IllegalStateException(s"Could not find broker with id $brokerId"))
  }
}


Review comment (suggestion):
```scala
IntegrationTestUtils.connectAndReceive[T](request, cluster.boundPorts().get(0))
```

Code after:
          .setGroupId(groupIds(i))
          .setErrorCode(expectedErrors(i).code)
      }.asJava.iterator))

    val deleteGroupsResponse = connectAndReceive[DeleteGroupsResponse](deleteGroupsRequest)
    assertEquals(expectedResponseData.results.asScala.toSet, deleteGroupsResponse.data.results.asScala.toSet)
  }

  protected def connectAny(): Socket = {
    val socket: Socket = IntegrationTestUtils.connect(cluster.boundPorts().get(0))
    openSockets += socket
    socket
  }

  protected def connect(destination: Int): Socket = {
    val socket = IntegrationTestUtils.connect(brokerSocketServer(destination).boundPort(cluster.clientListener()))
    openSockets += socket
    socket
  }

  protected def connectAndReceive[T <: AbstractResponse](
    request: AbstractRequest
  ): T = {
    IntegrationTestUtils.connectAndReceive[T](request, cluster.boundPorts().get(0))
  }

  protected def connectAndReceive[T <: AbstractResponse](
    request: AbstractRequest,
    destination: Int
  ): T = {
    val socketServer = brokerSocketServer(destination)
    val listenerName = cluster.clientListener()
    IntegrationTestUtils.connectAndReceive[T](request, socketServer.boundPort(listenerName))
  }

  private def brokerSocketServer(brokerId: Int): SocketServer = {
    getBrokers.find { broker =>
      broker.config.brokerId == brokerId
    }.map(_.socketServer).getOrElse(throw new IllegalStateException(s"Could not find broker with id $brokerId"))
  }
}