address comments

Author: TaiJuWu

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

@@ -605,7 +605,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
       // check controller listener names (they won't appear in listeners when process.roles=broker)
       // as well as listeners for occurrences of SSL or SASL_*
       if (controllerListenerNames.exists(isSslOrSasl) ||
-        Csv.parseCsvList(getString(SocketServerConfigs.LISTENERS_CONFIG)).asScala.exists(listenerValue => isSslOrSasl(EndPoint.parseListenerName(listenerValue)))) {
+        Csv.parseCsvList(getString(SocketServerConfigs.LISTENERS_CONFIG)).asScala.exists(listenerValue => isSslOrSasl(CoreUtils.parseListenerName(listenerValue)))) {
         mapValue // don't add default mappings since we found something that is SSL or SASL_*
       } else {
         // add the PLAINTEXT mappings for all controller listener names that are not explicitly PLAINTEXT

core/src/main/scala/kafka/utils/CoreUtils.scala

@@ -32,10 +32,12 @@ import org.apache.commons.validator.routines.InetAddressValidator
 import org.apache.kafka.common.network.ListenerName
 import org.apache.kafka.common.security.auth.SecurityProtocol
 import org.apache.kafka.common.utils.Utils
+import org.apache.kafka.common.KafkaException
 import org.apache.kafka.network.SocketServerConfigs
 import org.slf4j.event.Level
 
 import java.util
+import java.util.Locale
 import scala.jdk.CollectionConverters._
 
 /**
@@ -244,4 +246,12 @@ object CoreUtils {
   def replicaToBrokerAssignmentAsScala(map: util.Map[Integer, util.List[Integer]]): Map[Int, Seq[Int]] = {
     map.asScala.map(e => (e._1.asInstanceOf[Int], e._2.asScala.map(_.asInstanceOf[Int])))
   }
+
+  def parseListenerName(connectionString: String): String = {
+    val firstColon = connectionString.indexOf(':')
+    if (firstColon < 0) {
+      throw new KafkaException(s"Unable to parse a listener name from $connectionString")
+    }
+    connectionString.substring(0, firstColon).toUpperCase(Locale.ROOT)
+  }
 }

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

@@ -46,6 +46,8 @@ import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.function.Executable
 
+import org.apache.kafka.common.test.{TestUtils => JTestUtils}
+
 import scala.jdk.CollectionConverters._
 
 class KafkaConfigTest {
@@ -594,8 +596,8 @@ class KafkaConfigTest {
     props.setProperty(SocketServerConfigs.LISTENERS_CONFIG, "plaintext://localhost:9091,SsL://localhost:9092")
     props.setProperty(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "PLAINTEXT:PLAINTEXT,SSL:SSL,CONTROLLER:PLAINTEXT")
     val config = KafkaConfig.fromProps(props)
-    assertEquals(Some("SSL://localhost:9092"), config.listeners.find(_.listenerName.value == "SSL").map(_.connectionString))
-    assertEquals(Some("PLAINTEXT://localhost:9091"), config.listeners.find(_.listenerName.value == "PLAINTEXT").map(_.connectionString))
+    assertEquals(Some("SSL://localhost:9092"), config.listeners.find(_.listenerName.value == "SSL").map(JTestUtils.endpointToString))
+    assertEquals(Some("PLAINTEXT://localhost:9091"), config.listeners.find(_.listenerName.value == "PLAINTEXT").map(JTestUtils.endpointToString))
   }
 
   private def listenerListToEndPoints(listenerList: String,
@@ -1180,7 +1182,7 @@ class KafkaConfigTest {
 
     val config = KafkaConfig.fromProps(defaults)
     assertEquals(1, config.brokerId)
-    assertEquals(Seq("PLAINTEXT://127.0.0.1:1122"), config.effectiveAdvertisedBrokerListeners.map(_.connectionString))
+    assertEquals(Seq("PLAINTEXT://127.0.0.1:1122"), config.effectiveAdvertisedBrokerListeners.map(JTestUtils.endpointToString))
     assertEquals(Map("127.0.0.1" -> 2, "127.0.0.2" -> 3), config.maxConnectionsPerIpOverrides)
     assertEquals(List("/tmp1", "/tmp2"), config.logDirs)
     assertEquals(12 * 60L * 1000L * 60, config.logRollTimeMillis)

server/src/main/java/org/apache/kafka/network/EndPoint.java

@@ -17,12 +17,9 @@
 
 package org.apache.kafka.network;
 
-import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
-import org.apache.kafka.common.utils.Utils;
 
-import java.util.Locale;
 import java.util.Objects;
 
 public class EndPoint {
@@ -38,24 +35,11 @@ public EndPoint(String host, int port, ListenerName listenerName, SecurityProtoc
         this.securityProtocol = securityProtocol;
     }
 
-    public static String parseListenerName(String connectionString) {
-        int firstColon = connectionString.indexOf(':');
-        if (firstColon < 0) {
-            throw new KafkaException("Unable to parse a listener name from " + connectionString);
-        }
-        return connectionString.substring(0, firstColon).toUpperCase(Locale.ROOT);
-    }
-
     public static EndPoint fromPublic(org.apache.kafka.common.Endpoint endpoint) {
         return new EndPoint(endpoint.host(), endpoint.port(),
                 new ListenerName(endpoint.listenerName().get()), endpoint.securityProtocol());
     }
 
-    public String connectionString() {
-        String hostport = (host == null) ? (":" + port) : Utils.formatAddress(host, port);
-        return listenerName.value() + "://" + hostport;
-    }
-
     public org.apache.kafka.common.Endpoint toPublic() {
         return new org.apache.kafka.common.Endpoint(listenerName.value(), securityProtocol, host, port);
     }

test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java

@@ -21,8 +21,10 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.network.ListenerName;
 import org.apache.kafka.common.utils.Exit;
 import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.network.EndPoint;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -101,6 +103,18 @@ static File tempDirectory() {
         return file;
     }
 
+    /**
+     * Convert EndPoint to String
+     */
+    public static String endpointToString(EndPoint endPoint) {
+        String host = endPoint.host();
+        int port = endPoint.port();
+        ListenerName listenerName = endPoint.listenerName();
+
+        String hostport = (host == null) ? (":" + port) : Utils.formatAddress(host, port);
+        return listenerName.value() + "://" + hostport;
+    }
+
     /**
      * uses default value of 15 seconds for timeout
      */