KAFKA-18432: Remove Option wrapper from channelManager parameter

The `channelManager` parameter is now required and no longer optional.
apache · Jan 8, 2025 · f8ffab5 · f8ffab5
1 parent f530cd0
commit f8ffab5
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 9 deletions.
diff --git a/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala b/core/src/main/scala/kafka/server/AutoTopicCreationManager.scala
@@ -49,14 +49,11 @@ trait AutoTopicCreationManager {
 
 class DefaultAutoTopicCreationManager(
   config: KafkaConfig,
-  channelManager: Option[NodeToControllerChannelManager],
+  channelManager: NodeToControllerChannelManager,
   groupCoordinator: GroupCoordinator,
   txnCoordinator: TransactionCoordinator,
   shareCoordinator: Option[ShareCoordinator]
 ) extends AutoTopicCreationManager with Logging {
-  if (channelManager.isEmpty) {
-    throw new IllegalArgumentException("Must supply a channel manager for auto topic creation")
-  }
 
   private val inflightTopics = Collections.newSetFromMap(new ConcurrentHashMap[String, java.lang.Boolean]())
 
@@ -117,10 +114,6 @@ class DefaultAutoTopicCreationManager(
       }
     }
 
-    val channelManager = this.channelManager.getOrElse {
-      throw new IllegalStateException("Channel manager must be defined in order to send CreateTopic requests.")
-    }
-
     val request = metadataRequestContext.map { context =>
       val requestVersion =
         channelManager.controllerApiVersions.toScala match {

diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -385,7 +385,7 @@ class BrokerServer(
         producerIdManagerSupplier, metrics, metadataCache, Time.SYSTEM)
 
       autoTopicCreationManager = new DefaultAutoTopicCreationManager(
-        config, Some(clientToControllerChannelManager), groupCoordinator,
+        config, clientToControllerChannelManager, groupCoordinator,
         transactionCoordinator, shareCoordinator)
 
       dynamicConfigHandlers = Map[String, ConfigHandler](