* kafka: updated to 4.0.0

Signed-off-by: neo <[email protected]>

Author: neowu

CHANGELOG.md

@@ -8,6 +8,7 @@
 * sse: log event count and size on sse:close action
 * ws: remove ws support
   > websocket is not used anymore, use sse/ajax instead
+* kafka: updated to 4.0.0
 
 ### 9.1.7 (2/26/2025 - 3/6/2025)
 

build.gradle.kts

@@ -32,7 +32,7 @@ project("core-ng") {
         implementation("com.squareup.okio:okio:3.2.0")              // okio 3.3.0 has synchronization issue with virtual thread
         implementation("org.jetbrains.kotlin:kotlin-stdlib:2.0.20")
         implementation("io.undertow:undertow-core:2.3.18.Final")
-        implementation("org.apache.kafka:kafka-clients:3.9.0")
+        implementation("org.apache.kafka:kafka-clients:4.0.0")
         compileOnly("org.jboss.logging:jboss-logging-annotations:2.2.1.Final")
         compileOnly("com.github.spotbugs:spotbugs-annotations:4.8.3")
         testImplementation("org.junit.jupiter:junit-jupiter-api:${junitVersion}")

core-ng/src/main/java/core/framework/internal/kafka/EmptyMetricsReporter.java

@@ -0,0 +1,29 @@
+package core.framework.internal.kafka;
+
+import org.apache.kafka.common.metrics.KafkaMetric;
+import org.apache.kafka.common.metrics.MetricsReporter;
+
+import java.util.List;
+import java.util.Map;
+
+public class EmptyMetricsReporter implements MetricsReporter {
+    @Override
+    public void init(List<KafkaMetric> metrics) {
+    }
+
+    @Override
+    public void metricChange(KafkaMetric metric) {
+    }
+
+    @Override
+    public void metricRemoval(KafkaMetric metric) {
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+    }
+}

core-ng/src/main/java/core/framework/internal/kafka/MessageListener.java

@@ -97,7 +97,6 @@ String threadName(String name) {
         return "kafka-listener" + (name == null ? "" : "-" + name);
     }
 
-    @SuppressWarnings("deprecation")
     Consumer<String, byte[]> createConsumer() {
         var watch = new StopWatch();
         try {
@@ -114,7 +113,7 @@ Consumer<String, byte[]> createConsumer() {
             config.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, maxPollBytes);
             config.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, minPollBytes);
             config.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, (int) maxWaitTime.toMillis());
-            config.put(ConsumerConfig.AUTO_INCLUDE_JMX_REPORTER_CONFIG, Boolean.FALSE);
+            config.put(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, EmptyMetricsReporter.class.getName());
             config.put(ConsumerConfig.ENABLE_METRICS_PUSH_CONFIG, Boolean.FALSE);
             Consumer<String, byte[]> consumer = new KafkaConsumer<>(config, new KeyDeserializer(), new ByteArrayDeserializer());
             consumerMetrics.add(consumer.metrics());

core-ng/src/main/java/core/framework/internal/kafka/MessageProducer.java

@@ -43,7 +43,6 @@ public void send(ProducerRecord<byte[], byte[]> record) {
         producer.send(record, new KafkaCallback(record));
     }
 
-    @SuppressWarnings("deprecation")
     Producer<byte[], byte[]> createProducer(KafkaURI uri) {
         var watch = new StopWatch();
         try {
@@ -55,7 +54,7 @@ Producer<byte[], byte[]> createProducer(KafkaURI uri) {
                 ProducerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 5_000L,                      // 5s
                 ProducerConfig.MAX_BLOCK_MS_CONFIG, 30_000L,                                 // 30s, metadata update timeout, shorter than default, to get exception sooner if kafka is not available
                 ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize,
-                ProducerConfig.AUTO_INCLUDE_JMX_REPORTER_CONFIG, Boolean.FALSE,
+                ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, EmptyMetricsReporter.class.getName(),
                 ProducerConfig.ENABLE_METRICS_PUSH_CONFIG, Boolean.FALSE);
 
             var serializer = new ByteArraySerializer();

core-ng/src/main/java/core/framework/internal/log/DefaultLoggerFactory.java

@@ -23,10 +23,7 @@ private Logger createLogger(String name) {
 
     private LogLevel infoLevel(String name) {
         // kafka log info for every producer/consumer, to reduce verbosity
-        if ("org.apache.kafka.clients.consumer.ConsumerConfig".equals(name)
-                || "org.apache.kafka.clients.producer.ProducerConfig".equals(name)
-                || "org.apache.kafka.clients.admin.AdminClientConfig".equals(name)
-                || "org.apache.kafka.common.utils.AppInfoParser".equals(name)) {
+        if (name.startsWith("org.apache.kafka.")) {
             return LogLevel.WARN;
         }
         return LogLevel.INFO;

docker/kafka/docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   kafka:
-    image: apache/kafka:3.9.0
+    image: apache/kafka-native:4.0.0
     hostname: dev.internal
     ports:
     - 9092:9092
@@ -26,7 +26,7 @@ services:
     volumes:
     - data:/var/lib/kafka/data
 #  mm:
-#    image: apache/kafka:3.9.0
+#    image: apache/kafka:4.0.0
 #    volumes:
 #    - ./connect-mirror-maker.properties:/opt/kafka/config/connect-mirror-maker.properties
 #    entrypoint: ["/bin/bash", "-c", "/opt/kafka/bin/connect-mirror-maker.sh /opt/kafka/config/connect-mirror-maker.properties"]