Kafka 4.1.0 removes kafka.cluster.EndPoint

Author: ozangunalp

smallrye-reactive-messaging-kafka-test-companion/src/main/java/io/smallrye/reactive/messaging/kafka/companion/test/EmbeddedKafkaBroker.java

@@ -40,7 +40,6 @@
 import org.apache.kafka.storage.internals.log.CleanerConfig;
 import org.jboss.logging.Logger;
 
-import kafka.cluster.EndPoint;
 import kafka.server.KafkaConfig;
 import kafka.server.KafkaRaftServer;
 import scala.jdk.javaapi.StreamConverters;
@@ -254,7 +253,7 @@ public String getAdvertisedListeners() {
     }
 
     public List<String> getLogDirs() {
-        return getLogDirs(config);
+        return config.logDirs();
     }
 
     public int getNodeId() {
@@ -283,7 +282,7 @@ public static Endpoint endpoint(String listener, SecurityProtocol protocol, Stri
 
     public static Endpoint parseEndpoint(SecurityProtocol protocol, String listenerStr) {
         Endpoint endpoint = parseEndpoint(listenerStr);
-        return new Endpoint(endpoint.listenerName().orElse(protocol.name), protocol, endpoint.host(), endpoint.port());
+        return new Endpoint(listenerName(endpoint), protocol, endpoint.host(), endpoint.port());
     }
 
     public static Endpoint parseEndpoint(String listenerStr) {
@@ -311,7 +310,7 @@ public static Properties createDefaultBrokerConfig(int nodeId, Endpoint controll
 
         // Configure listeners
         Map<String, Endpoint> listeners = advertisedListeners.stream()
-                .map(l -> new Endpoint(l.listenerName().orElse(null), l.securityProtocol(), "", l.port()))
+                .map(l -> new Endpoint(l.listener(), l.securityProtocol(), "", l.port()))
                 .collect(Collectors.toMap(EmbeddedKafkaBroker::listenerName, Function.identity()));
         listeners.put(listenerName(controller), controller);
         listeners.put(listenerName(internalEndpoint), internalEndpoint);
@@ -372,7 +371,7 @@ public static KafkaConfig formatStorageFromConfig(Properties properties, String
                 .setNodeId(config.nodeId())
                 .setUnstableFeatureVersionsEnabled(config.unstableFeatureVersionsEnabled())
                 .setIgnoreFormatted(ignoreFormatted)
-                .setControllerListenerName(config.controllerListenerNames().head())
+                .setControllerListenerName(config.controllerListenerNames().get(0))
                 .setMetadataLogDirectory(config.metadataLogDir());
         configToLogDirectories(config).forEach(formatter::addDirectory);
         try {
@@ -384,8 +383,7 @@ public static KafkaConfig formatStorageFromConfig(Properties properties, String
     }
 
     static Set<String> configToLogDirectories(KafkaConfig config) {
-        TreeSet<String> dirs = new TreeSet<>();
-        config.logDirs().foreach(dirs::add);
+        TreeSet<String> dirs = new TreeSet<>(config.logDirs());
         String metadataLogDir = config.metadataLogDir();
         if (metadataLogDir != null) {
             dirs.add(metadataLogDir);
@@ -422,15 +420,10 @@ public static KafkaRaftServer createServer(final KafkaConfig config) {
 
     private static String getAdvertisedListeners(KafkaConfig config) {
         return StreamConverters.asJavaParStream(config.effectiveAdvertisedBrokerListeners())
-                .map(EndPoint::connectionString)
+                .map(EmbeddedKafkaBroker::toListenerString)
                 .collect(Collectors.joining(","));
     }
 
-    private static List<String> getLogDirs(KafkaConfig config) {
-        return StreamConverters.asJavaParStream(config.logDirs())
-                .collect(Collectors.toList());
-    }
-
     private static int getUnusedPort(int port) {
         if (port != 0) {
             return port;
@@ -464,7 +457,11 @@ private static String toProtocolMap(Endpoint endpoint) {
     }
 
     private static String listenerName(Endpoint endpoint) {
-        return endpoint.listenerName().orElse(endpoint.securityProtocol().name);
+        String listener = endpoint.listener();
+        if (listener != null && !listener.isBlank()) {
+            return listener;
+        }
+        return endpoint.securityProtocol().name;
     }
 
     public static class LoggingOutputStream extends java.io.OutputStream {