KAFKA-18373: Remove ZkMetadataCache

Signed-off-by: PoAn Yang <[email protected]>

Author: FrankYang0529

Diff for: core/src/main/scala/kafka/cluster/Partition.scala

@@ -23,7 +23,7 @@ import kafka.controller.{KafkaController, StateChangeLogger}
 import kafka.log._
 import kafka.log.remote.RemoteLogManager
 import kafka.server._
-import kafka.server.metadata.{KRaftMetadataCache, ZkMetadataCache}
+import kafka.server.metadata.KRaftMetadataCache
 import kafka.server.share.DelayedShareFetch
 import kafka.utils.CoreUtils.{inReadLock, inWriteLock}
 import kafka.utils._
@@ -1100,11 +1100,6 @@ class Partition(val topicPartition: TopicPartition,
           !kRaftMetadataCache.isBrokerShuttingDown(followerReplicaId) &&
           isBrokerEpochIsrEligible(storedBrokerEpoch, cachedBrokerEpoch)
 
-      // In ZK mode, we just ensure the broker is alive. Although we do not check for shutting down brokers here,
-      // the controller will block them from being added to ISR.
-      case zkMetadataCache: ZkMetadataCache =>
-        zkMetadataCache.hasAliveBroker(followerReplicaId)
-
       case _ => true
     }
   }

Diff for: core/src/main/scala/kafka/server/KafkaApis.scala

@@ -2369,31 +2369,6 @@ class KafkaApis(val requestChannel: RequestChannel,
         describeClientQuotasRequest.getErrorResponse(requestThrottleMs, Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
     } else {
       metadataSupport match {
-        case ZkSupport(adminManager, controller, zkClient, forwardingManager, metadataCache, _) =>
-          val result = adminManager.describeClientQuotas(describeClientQuotasRequest.filter)
-
-          val entriesData = result.iterator.map { case (quotaEntity, quotaValues) =>
-            val entityData = quotaEntity.entries.asScala.iterator.map { case (entityType, entityName) =>
-              new DescribeClientQuotasResponseData.EntityData()
-                .setEntityType(entityType)
-                .setEntityName(entityName)
-            }.toBuffer
-
-            val valueData = quotaValues.iterator.map { case (key, value) =>
-              new DescribeClientQuotasResponseData.ValueData()
-                .setKey(key)
-                .setValue(value)
-            }.toBuffer
-
-            new DescribeClientQuotasResponseData.EntryData()
-              .setEntity(entityData.asJava)
-              .setValues(valueData.asJava)
-          }.toBuffer
-
-          requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs =>
-            new DescribeClientQuotasResponse(new DescribeClientQuotasResponseData()
-              .setThrottleTimeMs(requestThrottleMs)
-              .setEntries(entriesData.asJava)))
         case RaftSupport(_, metadataCache) =>
           val result = metadataCache.describeClientQuotas(describeClientQuotasRequest.data())
           requestHelper.sendResponseMaybeThrottle(request, requestThrottleMs => {

Diff for: core/src/main/scala/kafka/server/MetadataCache.scala

@@ -17,12 +17,11 @@
 
 package kafka.server
 
-import kafka.server.metadata.{KRaftMetadataCache, ZkMetadataCache}
+import kafka.server.metadata.KRaftMetadataCache
 import org.apache.kafka.admin.BrokerMetadata
 import org.apache.kafka.common.message.{MetadataResponseData, UpdateMetadataRequestData}
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.{Cluster, Node, TopicPartition, Uuid}
-import org.apache.kafka.server.BrokerFeatures
 import org.apache.kafka.server.common.{FinalizedFeatures, KRaftVersion, MetadataVersion}
 
 import java.util
@@ -117,13 +116,6 @@ trait MetadataCache {
 }
 
 object MetadataCache {
-  def zkMetadataCache(brokerId: Int,
-                      metadataVersion: MetadataVersion,
-                      brokerFeatures: BrokerFeatures = BrokerFeatures.createEmpty())
-  : ZkMetadataCache = {
-    new ZkMetadataCache(brokerId, metadataVersion, brokerFeatures)
-  }
-
   def kRaftMetadataCache(
     brokerId: Int,
     kraftVersionSupplier: Supplier[KRaftVersion]

Diff for: core/src/main/scala/kafka/server/MetadataSupport.scala

@@ -17,10 +17,8 @@
 
 package kafka.server
 
-import kafka.controller.KafkaController
 import kafka.network.RequestChannel
-import kafka.server.metadata.{KRaftMetadataCache, ZkMetadataCache}
-import kafka.zk.KafkaZkClient
+import kafka.server.metadata.KRaftMetadataCache
 import org.apache.kafka.common.requests.AbstractResponse
 
 sealed trait MetadataSupport {
@@ -30,15 +28,6 @@ sealed trait MetadataSupport {
    */
   val forwardingManager: Option[ForwardingManager]
 
-  /**
-   * Return this instance downcast for use with ZooKeeper
-   *
-   * @param createException function to create an exception to throw
-   * @return this instance downcast for use with ZooKeeper
-   * @throws Exception if this instance is not for ZooKeeper
-   */
-  def requireZkOrThrow(createException: => Exception): ZkSupport
-
   /**
    * Return this instance downcast for use with Raft
    *
@@ -66,34 +55,10 @@ sealed trait MetadataSupport {
   }
 }
 
-case class ZkSupport(adminManager: ZkAdminManager,
-                     controller: KafkaController,
-                     zkClient: KafkaZkClient,
-                     forwardingManager: Option[ForwardingManager],
-                     metadataCache: ZkMetadataCache,
-                     brokerEpochManager: ZkBrokerEpochManager) extends MetadataSupport {
-  override def requireZkOrThrow(createException: => Exception): ZkSupport = this
-
-  override def requireRaftOrThrow(createException: => Exception): RaftSupport = throw createException
-
-  override def ensureConsistentWith(config: KafkaConfig): Unit = {
-    if (!config.requiresZookeeper) {
-      throw new IllegalStateException("Config specifies Raft but metadata support instance is for ZooKeeper")
-    }
-  }
-
-  override def canForward(): Boolean = forwardingManager.isDefined && (!controller.isActive)
-
-  def isBrokerEpochStale(brokerEpochInRequest: Long, isKRaftControllerRequest: Boolean): Boolean = {
-    brokerEpochManager.isBrokerEpochStale(brokerEpochInRequest, isKRaftControllerRequest)
-  }
-}
-
 case class RaftSupport(fwdMgr: ForwardingManager,
                        metadataCache: KRaftMetadataCache)
     extends MetadataSupport {
   override val forwardingManager: Option[ForwardingManager] = Some(fwdMgr)
-  override def requireZkOrThrow(createException: => Exception): ZkSupport = throw createException
   override def requireRaftOrThrow(createException: => Exception): RaftSupport = this
 
   override def ensureConsistentWith(config: KafkaConfig): Unit = {