KAFKA-13228: ApiVersionRequest is not properly handled in KRaft co-resident mode

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -98,7 +98,7 @@ public enum ApiKeys {
     END_QUORUM_EPOCH(ApiMessageType.END_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false),
     DESCRIBE_QUORUM(ApiMessageType.DESCRIBE_QUORUM, true, RecordBatch.MAGIC_VALUE_V0, true),
     ALTER_ISR(ApiMessageType.ALTER_ISR, true),
-    UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES, false, true),
+    UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES),
     ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false),
     FETCH_SNAPSHOT(ApiMessageType.FETCH_SNAPSHOT, false, RecordBatch.MAGIC_VALUE_V0, false),
     DESCRIBE_CLUSTER(ApiMessageType.DESCRIBE_CLUSTER),
@@ -267,6 +267,14 @@ public static EnumSet<ApiKeys> zkBrokerApis() {
         return apisForListener(ApiMessageType.ListenerType.ZK_BROKER);
     }
 
+    public static EnumSet<ApiKeys> controllerApis() {
+        return apisForListener(ApiMessageType.ListenerType.CONTROLLER);
+    }
+
+    public static EnumSet<ApiKeys> brokerApis() {
+        return apisForListener(ApiMessageType.ListenerType.BROKER);
+    }
+
     public static EnumSet<ApiKeys> apisForListener(ApiMessageType.ListenerType listener) {
         return APIS_BY_LISTENER.get(listener);
     }

core/src/main/scala/kafka/server/AlterIsrManager.scala

@@ -27,7 +27,7 @@ import kafka.zk.KafkaZkClient
 import org.apache.kafka.clients.ClientResponse
 import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.errors.OperationNotAttemptedException
-import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData}
+import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData, ApiMessageType}
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.protocol.Errors
 import org.apache.kafka.common.requests.{AlterIsrRequest, AlterIsrResponse}
@@ -86,7 +86,8 @@ object AlterIsrManager {
       config = config,
       channelName = "alterIsr",
       threadNamePrefix = threadNamePrefix,
-      retryTimeoutMs = Long.MaxValue
+      retryTimeoutMs = Long.MaxValue,
+      ApiMessageType.ListenerType.ZK_BROKER
     )
     new DefaultAlterIsrManager(
       controllerChannelManager = channelManager,

core/src/main/scala/kafka/server/BrokerServer.scala

@@ -218,7 +218,8 @@ class BrokerServer(
         config,
         channelName = "forwarding",
         threadNamePrefix,
-        retryTimeoutMs = 60000
+        retryTimeoutMs = 60000,
+        ListenerType.BROKER
       )
       clientToControllerChannelManager.start()
       forwardingManager = new ForwardingManagerImpl(clientToControllerChannelManager)
@@ -246,7 +247,8 @@ class BrokerServer(
         config,
         channelName = "alterIsr",
         threadNamePrefix,
-        retryTimeoutMs = Long.MaxValue
+        retryTimeoutMs = Long.MaxValue,
+        ListenerType.BROKER
       )
       alterIsrManager = new DefaultAlterIsrManager(
         controllerChannelManager = alterIsrChannelManager,
@@ -333,7 +335,7 @@ class BrokerServer(
       }
       lifecycleManager.start(() => metadataListener.highestMetadataOffset,
         BrokerToControllerChannelManager(controllerNodeProvider, time, metrics, config,
-          "heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong),
+          "heartbeat", threadNamePrefix, config.brokerSessionTimeoutMs.toLong, ListenerType.BROKER),
         metaProps.clusterId, networkListeners, supportedFeatures)
 
       // Register a listener with the Raft layer to receive metadata event notifications

core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala

@@ -19,16 +19,16 @@ package kafka.server
 
 import java.util.concurrent.LinkedBlockingDeque
 import java.util.concurrent.atomic.AtomicReference
-
 import kafka.common.{InterBrokerSendThread, RequestAndCompletionHandler}
 import kafka.raft.RaftManager
 import kafka.utils.Logging
 import org.apache.kafka.clients._
 import org.apache.kafka.common.Node
+import org.apache.kafka.common.message.ApiMessageType
 import org.apache.kafka.common.metrics.Metrics
 import org.apache.kafka.common.network._
-import org.apache.kafka.common.protocol.Errors
-import org.apache.kafka.common.requests.AbstractRequest
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
+import org.apache.kafka.common.requests.{AbstractRequest, ApiVersionsResponse}
 import org.apache.kafka.common.security.JaasContext
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.{LogContext, Time}
@@ -119,7 +119,8 @@ object BrokerToControllerChannelManager {
     config: KafkaConfig,
     channelName: String,
     threadNamePrefix: Option[String],
-    retryTimeoutMs: Long
+    retryTimeoutMs: Long,
+    controllerBrokerType: ApiMessageType.ListenerType
   ): BrokerToControllerChannelManager = {
     new BrokerToControllerChannelManagerImpl(
       controllerNodeProvider,
@@ -128,7 +129,8 @@ object BrokerToControllerChannelManager {
       config,
       channelName,
       threadNamePrefix,
-      retryTimeoutMs
+      retryTimeoutMs,
+      controllerBrokerType
     )
   }
 }
@@ -159,12 +161,13 @@ class BrokerToControllerChannelManagerImpl(
   config: KafkaConfig,
   channelName: String,
   threadNamePrefix: Option[String],
-  retryTimeoutMs: Long
+  retryTimeoutMs: Long,
+  controllerBrokerType: ApiMessageType.ListenerType
 ) extends BrokerToControllerChannelManager with Logging {
   private val logContext = new LogContext(s"[BrokerToControllerChannelManager broker=${config.brokerId} name=$channelName] ")
   private val manualMetadataUpdater = new ManualMetadataUpdater()
   private val apiVersions = new ApiVersions()
-  private val currentNodeApiVersions = NodeApiVersions.create()
+  private val currentNodeControllerApiVersions = new NodeApiVersions(ApiKeys.apisForListener(controllerBrokerType).asScala.map(ApiVersionsResponse.toApiVersion).asJava)
   private val requestThread = newRequestThread
 
   def start(): Unit = {
@@ -253,7 +256,7 @@ class BrokerToControllerChannelManagerImpl(
   def controllerApiVersions(): Option[NodeApiVersions] =
     requestThread.activeControllerAddress().flatMap(
       activeController => if (activeController.id() == config.brokerId)
-        Some(currentNodeApiVersions)
+        Some(currentNodeControllerApiVersions)
       else
         Option(apiVersions.get(activeController.idString()))
   )

core/src/main/scala/kafka/server/ControllerApis.scala

@@ -183,7 +183,7 @@ class ControllerApis(val requestChannel: RequestChannel,
         throw new TopicDeletionDisabledException()
       }
     }
-    val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS);
+    val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS)
     // The first step is to load up the names and IDs that have been provided by the
     // request.  This is a bit messy because we support multiple ways of referring to
     // topics (both by name and by id) and because we need to check for duplicates or
@@ -704,7 +704,7 @@ class ControllerApis(val requestChannel: RequestChannel,
                        hasClusterAuth: Boolean,
                        getCreatableTopics: Iterable[String] => Set[String])
                        : CompletableFuture[util.List[CreatePartitionsTopicResult]] = {
-    val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS);
+    val deadlineNs = time.nanoseconds() + NANOSECONDS.convert(request.timeoutMs, MILLISECONDS)
     val responses = new util.ArrayList[CreatePartitionsTopicResult]()
     val duplicateTopicNames = new util.HashSet[String]()
     val topicNames = new util.HashSet[String]()

core/src/main/scala/kafka/server/KafkaApis.scala

@@ -219,7 +219,7 @@ class KafkaApis(val requestChannel: RequestChannel,
         case ApiKeys.DESCRIBE_USER_SCRAM_CREDENTIALS => handleDescribeUserScramCredentialsRequest(request)
         case ApiKeys.ALTER_USER_SCRAM_CREDENTIALS => maybeForwardToController(request, handleAlterUserScramCredentialsRequest)
         case ApiKeys.ALTER_ISR => handleAlterIsrRequest(request)
-        case ApiKeys.UPDATE_FEATURES => maybeForwardToController(request, handleUpdateFeatures)
+        case ApiKeys.UPDATE_FEATURES => handleUpdateFeatures(request)
         case ApiKeys.ENVELOPE => handleEnvelope(request, requestLocal)
         case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request)
         case ApiKeys.DESCRIBE_PRODUCERS => handleDescribeProducersRequest(request)
@@ -3244,7 +3244,7 @@ class KafkaApis(val requestChannel: RequestChannel,
   }
 
   def handleUpdateFeatures(request: RequestChannel.Request): Unit = {
-    val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.shouldAlwaysForward(request))
+    val zkSupport = metadataSupport.requireZkOrThrow(KafkaApis.notYetSupported(request))
     val updateFeaturesRequest = request.body[UpdateFeaturesRequest]
 
     def sendResponseCallback(errors: Either[ApiError, Map[String, ApiError]]): Unit = {

core/src/main/scala/kafka/server/KafkaServer.scala

@@ -269,7 +269,9 @@ class KafkaServer(
           config = config,
           channelName = "forwarding",
           threadNamePrefix = threadNamePrefix,
-          retryTimeoutMs = config.requestTimeoutMs.longValue)
+          retryTimeoutMs = config.requestTimeoutMs.longValue,
+          ListenerType.ZK_BROKER
+        )
         clientToControllerChannelManager.start()
 
         /* start forwarding manager */

core/src/test/scala/unit/kafka/server/AbstractApiVersionsRequestTest.scala

@@ -16,13 +16,15 @@
  */
 package kafka.server
 
-import java.util.Properties
-
 import kafka.test.ClusterInstance
+import kafka.test.ClusterInstance.ClusterType
+import org.apache.kafka.clients.NodeApiVersions
 import org.apache.kafka.common.message.ApiMessageType.ListenerType
 import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion
+import org.apache.kafka.common.message.{ApiMessageType, ApiVersionsResponseData}
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.protocol.ApiKeys
+import org.apache.kafka.common.record.RecordVersion
 import org.apache.kafka.common.requests.{ApiVersionsRequest, ApiVersionsResponse, RequestUtils}
 import org.apache.kafka.common.utils.Utils
 import org.junit.jupiter.api.Assertions._
@@ -39,15 +41,6 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {
 
   def controlPlaneListenerName = new ListenerName("CONTROLLER")
 
-  // Configure control plane listener to make sure we have separate listeners for testing.
-  def brokerPropertyOverrides(properties: Properties): Unit = {
-    val securityProtocol = cluster.config().securityProtocol()
-    properties.setProperty(KafkaConfig.ControlPlaneListenerNameProp, controlPlaneListenerName.value())
-    properties.setProperty(KafkaConfig.ListenerSecurityProtocolMapProp, s"${controlPlaneListenerName.value()}:$securityProtocol,$securityProtocol:$securityProtocol")
-    properties.setProperty("listeners", s"$securityProtocol://localhost:0,${controlPlaneListenerName.value()}://localhost:0")
-    properties.setProperty(KafkaConfig.AdvertisedListenersProp, s"$securityProtocol://localhost:0,${controlPlaneListenerName.value()}://localhost:0")
-  }
-
   def sendUnsupportedApiVersionRequest(request: ApiVersionsRequest): ApiVersionsResponse = {
     val overrideHeader = IntegrationTestUtils.nextRequestHeader(ApiKeys.API_VERSIONS, Short.MaxValue)
     val socket = IntegrationTestUtils.connect(cluster.brokerSocketServers().asScala.head, cluster.clientListener())
@@ -60,14 +53,34 @@ abstract class AbstractApiVersionsRequestTest(cluster: ClusterInstance) {
   }
 
   def validateApiVersionsResponse(apiVersionsResponse: ApiVersionsResponse): Unit = {
-    val expectedApis = ApiKeys.zkBrokerApis()
+    val expectedApis = if (cluster.clusterType() == ClusterType.ZK)
+      ApiKeys.zkBrokerApis()
+    else {
+      val apis = ApiVersionsResponse.intersectForwardableApis(
+        ApiMessageType.ListenerType.BROKER,
+        RecordVersion.current,
+        new NodeApiVersions(ApiKeys.controllerApis().asScala.map(ApiVersionsResponse.toApiVersion).asJava).allSupportedApiVersions()
+      )
+      apis.add(
+        new ApiVersionsResponseData.ApiVersion()
+          .setApiKey(ApiKeys.ENVELOPE.id)
+          .setMinVersion(ApiKeys.ENVELOPE.oldestVersion)
+          .setMaxVersion(ApiKeys.ENVELOPE.latestVersion)
+      )
+      apis
+    }
+
     assertEquals(expectedApis.size(), apiVersionsResponse.data.apiKeys().size(),
       "API keys in ApiVersionsResponse must match API keys supported by broker.")
 
-    val defaultApiVersionsResponse = ApiVersionsResponse.defaultApiVersionsResponse(ListenerType.ZK_BROKER)
+    val defaultApiVersionsResponse = if (cluster.clusterType() == ClusterType.ZK)
+      ApiVersionsResponse.defaultApiVersionsResponse(ListenerType.ZK_BROKER)
+    else
+      ApiVersionsResponse.createApiVersionsResponse(0, expectedApis.asInstanceOf[ApiVersionsResponseData.ApiVersionCollection])
+
     for (expectedApiVersion: ApiVersion <- defaultApiVersionsResponse.data.apiKeys().asScala) {
       val actualApiVersion = apiVersionsResponse.apiVersion(expectedApiVersion.apiKey)
-      assertNotNull(actualApiVersion, s"API key ${actualApiVersion.apiKey} is supported by broker, but not received in ApiVersionsResponse.")
+      assertNotNull(actualApiVersion, s"API key ${expectedApiVersion.apiKey()} is supported by broker, but not received in ApiVersionsResponse.")
       assertEquals(expectedApiVersion.apiKey, actualApiVersion.apiKey, "API key must be supported by the broker.")
       assertEquals(expectedApiVersion.minVersion, actualApiVersion.minVersion, s"Received unexpected min version for API key ${actualApiVersion.apiKey}.")
       assertEquals(expectedApiVersion.maxVersion, actualApiVersion.maxVersion, s"Received unexpected max version for API key ${actualApiVersion.apiKey}.")

core/src/test/scala/unit/kafka/server/ApiVersionsRequestTest.scala

@@ -17,25 +17,20 @@
 
 package kafka.server
 
-import kafka.test.{ClusterConfig, ClusterInstance}
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
 import org.apache.kafka.common.message.ApiVersionsRequestData
 import org.apache.kafka.common.protocol.{ApiKeys, Errors}
 import org.apache.kafka.common.requests.ApiVersionsRequest
-import kafka.test.annotation.ClusterTest
-import kafka.test.junit.ClusterTestExtensions
 import org.junit.jupiter.api.Assertions._
-import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.extension.ExtendWith
 
 
 @ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(clusterType = Type.BOTH, brokers = 1)
 class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersionsRequestTest(cluster) {
 
-  @BeforeEach
-  def setup(config: ClusterConfig): Unit = {
-    super.brokerPropertyOverrides(config.serverProperties())
-  }
-
   @ClusterTest
   def testApiVersionsRequest(): Unit = {
     val request = new ApiVersionsRequest.Builder().build()
@@ -46,7 +41,7 @@ class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersio
   @ClusterTest
   def testApiVersionsRequestThroughControlPlaneListener(): Unit = {
     val request = new ApiVersionsRequest.Builder().build()
-    val apiVersionsResponse = sendApiVersionsRequest(request, super.controlPlaneListenerName)
+    val apiVersionsResponse = sendApiVersionsRequest(request, cluster.clientListener)
     validateApiVersionsResponse(apiVersionsResponse)
   }
 
@@ -72,7 +67,7 @@ class ApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVersio
   @ClusterTest
   def testApiVersionsRequestValidationV0ThroughControlPlaneListener(): Unit = {
     val apiVersionsRequest = new ApiVersionsRequest.Builder().build(0.asInstanceOf[Short])
-    val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest, super.controlPlaneListenerName)
+    val apiVersionsResponse = sendApiVersionsRequest(apiVersionsRequest, cluster.clientListener)
     validateApiVersionsResponse(apiVersionsResponse)
   }
 

core/src/test/scala/unit/kafka/server/KafkaApisTest.scala

@@ -893,12 +893,6 @@ class KafkaApisTest {
     testForwardableApi(ApiKeys.DELETE_TOPICS, requestBuilder)
   }
 
-  @Test
-  def testUpdateFeaturesWithForwarding(): Unit = {
-    val requestBuilder = new UpdateFeaturesRequest.Builder(new UpdateFeaturesRequestData())
-    testForwardableApi(ApiKeys.UPDATE_FEATURES, requestBuilder)
-  }
-
   @Test
   def testAlterScramWithForwarding(): Unit = {
     val requestBuilder = new AlterUserScramCredentialsRequest.Builder(new AlterUserScramCredentialsRequestData())
@@ -4137,12 +4131,6 @@ class KafkaApisTest {
     verifyShouldAlwaysForwardErrorMessage(createKafkaApis(raftSupport = true).handleAlterUserScramCredentialsRequest)
   }
 
-  @Test
-  def testRaftShouldAlwaysForwardUpdateFeatures(): Unit = {
-    metadataCache = MetadataCache.kRaftMetadataCache(brokerId)
-    verifyShouldAlwaysForwardErrorMessage(createKafkaApis(raftSupport = true).handleUpdateFeatures)
-  }
-
   @Test
   def testRaftShouldAlwaysForwardElectLeaders(): Unit = {
     metadataCache = MetadataCache.kRaftMetadataCache(brokerId)

core/src/test/scala/unit/kafka/server/SaslApiVersionsRequestTest.scala

@@ -49,7 +49,6 @@ class SaslApiVersionsRequestTest(cluster: ClusterInstance) extends AbstractApiVe
     sasl.startSasl(sasl.jaasSections(kafkaServerSaslMechanisms, Some(kafkaClientSaslMechanism), KafkaSasl, JaasTestUtils.KafkaServerContextName))
     config.saslServerProperties().putAll(sasl.kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism))
     config.saslClientProperties().putAll(sasl.kafkaClientSaslProperties(kafkaClientSaslMechanism))
-    super.brokerPropertyOverrides(config.serverProperties())
   }
 
   @ClusterTest(securityProtocol = SecurityProtocol.SASL_PLAINTEXT, clusterType = Type.ZK)

raft/src/main/java/org/apache/kafka/raft/RaftConfig.java

@@ -264,7 +264,7 @@ public void ensureValid(String name, Object value) {
             }
 
             @SuppressWarnings("unchecked")
-            List<String> voterStrings = (List) value;
+            List<String> voterStrings = (List<String>) value;
 
             // Attempt to parse the connect strings
             parseVoterConnections(voterStrings);