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

Code before:
import org.apache.kafka.server.IntegrationTestUtils
import org.apache.kafka.server.common.{EligibleLeaderReplicasVersion, GroupVersion, MetadataVersion, ShareVersion, StreamsVersion, TransactionVersion}
import org.apache.kafka.test.TestUtils
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Tag

import scala.jdk.CollectionConverters._

@Tag("integration")
abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {

  def sendApiVersionsRequest(request: ApiVersionsRequest, listenerName: ListenerName): ApiVersionsResponse = {
    val socket = if (cluster.controllerListenerName() == listenerName) {
      cluster.controllerSocketServers().asScala.head
    } else {
      cluster.brokerSocketServers().asScala.head
    }

    val port = socket.boundPort(listenerName)

    IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request, port)
  }

  def sendUnsupportedApiVersionRequest(request: ApiVersionsRequest): ApiVersionsResponse = {
    val overrideHeader = IntegrationTestUtils.nextRequestHeader(ApiKeys.API_VERSIONS, Short.MaxValue)
    val socketServer = cluster.brokerSocketServers().asScala.head
    val listenerName = cluster.clientListener()
    val port = socketServer.boundPort(listenerName)
    val socket = IntegrationTestUtils.connect(port)
    try {
      val serializedBytes = Utils.toArray(
        RequestUtils.serialize(overrideHeader.data, overrideHeader.headerVersion, request.data, request.version))
      IntegrationTestUtils.sendRequest(socket, serializedBytes)
      IntegrationTestUtils.receive[ApiVersionsResponse](socket, ApiKeys.API_VERSIONS, 0.toShort)
    } finally socket.close()
  }

  def validateApiVersionsResponse(
    apiVersionsResponse: ApiVersionsResponse,
    listenerName: ListenerName = cluster.clientListener(),
    enableUnstableLastVersion: Boolean = false,
    clientTelemetryEnabled: Boolean = false,
    apiVersion: Short = ApiKeys.API_VERSIONS.latestVersion
  ): Unit = {
    if (apiVersion >= 3) {
      assertEquals(6, apiVersionsResponse.data().finalizedFeatures().size())
      assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).minVersionLevel())
      assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).maxVersionLevel())

      assertEquals(7, apiVersionsResponse.data().supportedFeatures().size())
      assertEquals(MetadataVersion.MINIMUM_VERSION.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(MetadataVersion.FEATURE_NAME).minVersion())

Review comment (refactor):
Could you please add following helper to `ClusterInstance`?
```
    default Collection<Integer> boundPorts() {
        return brokers().values().stream()
                .map(KafkaBroker::socketServer)
                .map(s -> s.boundPort(clientListener()))
                .collect(Collectors.toList());
    }
```

this helper can streamline this PR I believe

Code after:
import org.apache.kafka.server.IntegrationTestUtils
import org.apache.kafka.server.common.{EligibleLeaderReplicasVersion, GroupVersion, MetadataVersion, ShareVersion, StreamsVersion, TransactionVersion}
import org.apache.kafka.test.TestUtils
import org.junit.jupiter.api.Assertions._
import org.junit.jupiter.api.Tag

import scala.jdk.CollectionConverters._

@Tag("integration")
abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {

  def sendApiVersionsRequest(request: ApiVersionsRequest, listenerName: ListenerName): ApiVersionsResponse = {
    val socket = if (cluster.controllerListenerName() == listenerName) {
      cluster.controllerSocketServers().asScala.head
    } else {
      cluster.brokerSocketServers().asScala.head
    }
    IntegrationTestUtils.connectAndReceive[ApiVersionsResponse](request, socket.boundPort(listenerName))
  }

  def sendUnsupportedApiVersionRequest(request: ApiVersionsRequest): ApiVersionsResponse = {
    val overrideHeader = IntegrationTestUtils.nextRequestHeader(ApiKeys.API_VERSIONS, Short.MaxValue)
    val socket = IntegrationTestUtils.connect(cluster.boundPorts().get(0))
    try {
      val serializedBytes = Utils.toArray(
        RequestUtils.serialize(overrideHeader.data, overrideHeader.headerVersion, request.data, request.version))
      IntegrationTestUtils.sendRequest(socket, serializedBytes)
      IntegrationTestUtils.receive[ApiVersionsResponse](socket, ApiKeys.API_VERSIONS, 0.toShort)
    } finally socket.close()
  }

  def validateApiVersionsResponse(
    apiVersionsResponse: ApiVersionsResponse,
    listenerName: ListenerName = cluster.clientListener(),
    enableUnstableLastVersion: Boolean = false,
    clientTelemetryEnabled: Boolean = false,
    apiVersion: Short = ApiKeys.API_VERSIONS.latestVersion
  ): Unit = {
    if (apiVersion >= 3) {
      assertEquals(6, apiVersionsResponse.data().finalizedFeatures().size())
      assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).minVersionLevel())
      assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().finalizedFeatures().find(MetadataVersion.FEATURE_NAME).maxVersionLevel())

      assertEquals(7, apiVersionsResponse.data().supportedFeatures().size())
      assertEquals(MetadataVersion.MINIMUM_VERSION.featureLevel(), apiVersionsResponse.data().supportedFeatures().find(MetadataVersion.FEATURE_NAME).minVersion())
      if (apiVersion < 4) {
        assertEquals(1, apiVersionsResponse.data().supportedFeatures().find("kraft.version").minVersion())
      } else {
        assertEquals(0, apiVersionsResponse.data().supportedFeatures().find("kraft.version").minVersion())
      }
      assertEquals(MetadataVersion.latestTesting().featureLevel(), apiVersionsResponse.data().supportedFeatures().find(MetadataVersion.FEATURE_NAME).maxVersion())