Bump kafka.version from 4.0.0 to 4.1.0 (#3141)

* Bump kafka.version from 4.0.0 to 4.1.0

Bumps `kafka.version` from 4.0.0 to 4.1.0.

Updates `org.apache.kafka:kafka-clients` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka_2.13` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-server-common` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-metadata` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-server` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-raft` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-group-coordinator` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-transaction-coordinator` from 4.0.0 to 4.1.0

Updates `org.apache.kafka:kafka-storage` from 4.0.0 to 4.1.0

---
updated-dependencies:
- dependency-name: org.apache.kafka:kafka-clients
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka_2.13
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-server-common
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-metadata
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-server
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-raft
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-group-coordinator
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-transaction-coordinator
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
- dependency-name: org.apache.kafka:kafka-storage
  dependency-version: 4.1.0
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <[email protected]>

* Kafka 4.1.0 removes kafka.cluster.EndPoint

---------

Signed-off-by: dependabot[bot] <[email protected]>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Ozan Gunalp <[email protected]>

Author: dependabot[bot]

pom.xml

@@ -87,7 +87,7 @@
 
     <jboss-log-manager.version>3.1.2.Final</jboss-log-manager.version>
 
-    <kafka.version>4.0.0</kafka.version>
+    <kafka.version>4.1.0</kafka.version>
 
     <opentelemetry.instrumentation.version>2.21.0</opentelemetry.instrumentation.version>
     <opentelemetry.version>1.55.0</opentelemetry.version>

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 {