[GitHub] [kafka] mumrah commented on a change in pull request #11070: Validate the controllerListener config on startup

[GitBox <gi...@apache.org>]

mumrah commented on a change in pull request #11070:
URL: https://github.com/apache/kafka/pull/11070#discussion_r672407194



##########
File path: core/src/main/scala/kafka/server/ControllerServer.scala
##########
@@ -137,8 +137,14 @@ class ControllerServer(
         credentialProvider,
         apiVersionManager)
       socketServer.startup(startProcessingRequests = false, controlPlaneListener = None, config.controllerListeners)
-      socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
-        config.controllerListeners.head.listenerName))
+
+      if(config.controllerListeners.nonEmpty) {
+        socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
+          config.controllerListeners.head.listenerName))
+      } else {
+        fatal("No controllerListener defined for controller")
+        throw new IllegalArgumentException()

Review comment:
       We have `ConfigException` for these such cases

##########
File path: core/src/main/scala/kafka/server/KafkaConfig.scala
##########
@@ -1914,6 +1914,8 @@ class KafkaConfig(val props: java.util.Map[_, _], doLog: Boolean, dynamicConfigO
       "offsets.commit.required.acks must be greater or equal -1 and less or equal to offsets.topic.replication.factor")
     require(BrokerCompressionCodec.isValid(compressionType), "compression.type : " + compressionType + " is not valid." +
       " Valid options are " + BrokerCompressionCodec.brokerCompressionOptions.mkString(","))
+    require(!processRoles.contains(ControllerRole) || controllerListeners.nonEmpty,
+      s"${KafkaConfig.ControllerListenerNamesProp} cannot be empty")

Review comment:
       How about "[...] cannot be empty if the server has the controller role"

##########
File path: core/src/main/scala/kafka/server/ControllerServer.scala
##########
@@ -137,8 +137,14 @@ class ControllerServer(
         credentialProvider,
         apiVersionManager)
       socketServer.startup(startProcessingRequests = false, controlPlaneListener = None, config.controllerListeners)
-      socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
-        config.controllerListeners.head.listenerName))
+
+      if(config.controllerListeners.nonEmpty) {
+        socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
+          config.controllerListeners.head.listenerName))
+      } else {
+        fatal("No controllerListener defined for controller")

Review comment:
       Instead of "controllerListener" let's log the full property name

##########
File path: core/src/main/scala/kafka/server/ControllerServer.scala
##########
@@ -137,8 +137,14 @@ class ControllerServer(
         credentialProvider,
         apiVersionManager)
       socketServer.startup(startProcessingRequests = false, controlPlaneListener = None, config.controllerListeners)
-      socketServerFirstBoundPortFuture.complete(socketServer.boundPort(
-        config.controllerListeners.head.listenerName))
+
+      if(config.controllerListeners.nonEmpty) {

Review comment:
       nit: need space after `if` 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org