KAFKA-12208: Rename AdminManager to ZkAdminManager (apache#9900)

Rename AdminManager to ZkAdminManager to emphasize the fact that it is not used by the KIP-500 code paths.

Reviewers: Ismael Juma <[email protected]>, Boyang Chen <[email protected]>, Chia-Ping Tsai <[email protected]>

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

@@ -47,7 +47,7 @@ object CreatePartitionsMetadata {
   */
 class DelayedCreatePartitions(delayMs: Long,
                               createMetadata: Seq[CreatePartitionsMetadata],
-                              adminManager: AdminManager,
+                              adminManager: ZkAdminManager,
                               responseCallback: Map[String, ApiError] => Unit)
   extends DelayedOperation(delayMs) {
 

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

@@ -38,7 +38,7 @@ object DeleteTopicMetadata {
   */
 class DelayedDeleteTopics(delayMs: Long,
                           deleteMetadata: Seq[DeleteTopicMetadata],
-                          adminManager: AdminManager,
+                          adminManager: ZkAdminManager,
                           responseCallback: Map[String, Errors] => Unit)
   extends DelayedOperation(delayMs) {
 

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

@@ -314,7 +314,7 @@ class DynamicBrokerConfig(private val kafkaConfig: KafkaConfig) extends Logging
    * All config updates through ZooKeeper are triggered through actual changes in values stored in ZooKeeper.
    * For some configs like SSL keystores and truststores, we also want to reload the store if it was modified
    * in-place, even though the actual value of the file path and password haven't changed. This scenario alone
-   * is handled here when a config update request using admin client is processed by AdminManager. If any of
+   * is handled here when a config update request using admin client is processed by ZkAdminManager. If any of
    * the SSL configs have changed, then the update will not be done here, but will be handled later when ZK
    * changes are processed. At the moment, only listener configs are considered for reloading.
    */

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

@@ -98,7 +98,7 @@ import scala.annotation.nowarn
  */
 class KafkaApis(val requestChannel: RequestChannel,
                 val replicaManager: ReplicaManager,
-                val adminManager: AdminManager,
+                val adminManager: ZkAdminManager,
                 val groupCoordinator: GroupCoordinator,
                 val txnCoordinator: TransactionCoordinator,
                 val controller: KafkaController,

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

@@ -154,7 +154,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
   var logManager: LogManager = null
 
   var replicaManager: ReplicaManager = null
-  var adminManager: AdminManager = null
+  var adminManager: ZkAdminManager = null
   var tokenManager: DelegationTokenManager = null
 
   var dynamicConfigHandlers: Map[String, ConfigHandler] = null
@@ -352,7 +352,7 @@ class KafkaServer(val config: KafkaConfig, time: Time = Time.SYSTEM, threadNameP
           forwardingManager.start()
         }
 
-        adminManager = new AdminManager(config, metrics, metadataCache, zkClient)
+        adminManager = new ZkAdminManager(config, metrics, metadataCache, zkClient)
 
         /* start group coordinator */
         // Hardcode Time.SYSTEM for now as some Streams tests fail otherwise, it would be good to fix the underlying issue

core/src/main/scala/kafka/server/AdminManager.scala → core/src/main/scala/kafka/server/ZkAdminManager.scala

@@ -57,7 +57,7 @@ import org.apache.kafka.common.utils.Sanitizer
 import scala.collection.{Map, mutable, _}
 import scala.jdk.CollectionConverters._
 
-class AdminManager(val config: KafkaConfig,
+class ZkAdminManager(val config: KafkaConfig,
                    val metrics: Metrics,
                    val metadataCache: MetadataCache,
                    val zkClient: KafkaZkClient) extends Logging with KafkaMetricsGroup {

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

@@ -83,7 +83,7 @@ class KafkaApisTest {
   private val requestChannelMetrics: RequestChannel.Metrics = EasyMock.createNiceMock(classOf[RequestChannel.Metrics])
   private val replicaManager: ReplicaManager = EasyMock.createNiceMock(classOf[ReplicaManager])
   private val groupCoordinator: GroupCoordinator = EasyMock.createNiceMock(classOf[GroupCoordinator])
-  private val adminManager: AdminManager = EasyMock.createNiceMock(classOf[AdminManager])
+  private val adminManager: ZkAdminManager = EasyMock.createNiceMock(classOf[ZkAdminManager])
   private val txnCoordinator: TransactionCoordinator = EasyMock.createNiceMock(classOf[TransactionCoordinator])
   private val controller: KafkaController = EasyMock.createNiceMock(classOf[KafkaController])
   private val forwardingManager: ForwardingManager = EasyMock.createNiceMock(classOf[ForwardingManager])

core/src/test/scala/unit/kafka/server/AdminManagerTest.scala → core/src/test/scala/unit/kafka/server/ZkAdminManagerTest.scala

@@ -33,7 +33,7 @@ import org.junit.Assert.assertNotNull
 import org.junit.Assert.assertNotEquals
 import java.util.Properties
 
-class AdminManagerTest {
+class ZkAdminManagerTest {
 
   private val zkClient: KafkaZkClient = EasyMock.createNiceMock(classOf[KafkaZkClient])
   private val metrics = new Metrics()
@@ -46,9 +46,9 @@ class AdminManagerTest {
     metrics.close()
   }
 
-  def createAdminManager(): AdminManager = {
+  def createAdminManager(): ZkAdminManager = {
     val props = TestUtils.createBrokerConfig(brokerId, "zk")
-    new AdminManager(KafkaConfig.fromProps(props), metrics, metadataCache, zkClient)
+    new ZkAdminManager(KafkaConfig.fromProps(props), metrics, metadataCache, zkClient)
   }
 
   @Test

jmh-benchmarks/src/main/java/org/apache/kafka/jmh/metadata/MetadataRequestBenchmark.java

@@ -22,7 +22,7 @@
 import kafka.coordinator.transaction.TransactionCoordinator;
 import kafka.network.RequestChannel;
 import kafka.network.RequestConvertToJson;
-import kafka.server.AdminManager;
+import kafka.server.ZkAdminManager;
 import kafka.server.BrokerFeatures;
 import kafka.server.BrokerTopicStats;
 import kafka.server.ClientQuotaManager;
@@ -96,7 +96,7 @@ public class MetadataRequestBenchmark {
     private RequestChannel.Metrics requestChannelMetrics = Mockito.mock(RequestChannel.Metrics.class);
     private ReplicaManager replicaManager = Mockito.mock(ReplicaManager.class);
     private GroupCoordinator groupCoordinator = Mockito.mock(GroupCoordinator.class);
-    private AdminManager adminManager = Mockito.mock(AdminManager.class);
+    private ZkAdminManager adminManager = Mockito.mock(ZkAdminManager.class);
     private TransactionCoordinator transactionCoordinator = Mockito.mock(TransactionCoordinator.class);
     private KafkaController kafkaController = Mockito.mock(KafkaController.class);
     private ForwardingManager forwardingManager = Mockito.mock(ForwardingManager.class);