[#10467] Bump spring-kafka from 2.9.11 to 3.0.12

pinpoint-apm · Nov 7, 2023 · 8d353a4 · 8d353a4
1 parent d8aa8c6
commit 8d353a4
Show file tree

Hide file tree

Showing 8 changed files with 38 additions and 38 deletions.
diff --git a/...main/java/com/navercorp/pinpoint/exceptiontrace/collector/dao/PinotExceptionTraceDao.java b/...main/java/com/navercorp/pinpoint/exceptiontrace/collector/dao/PinotExceptionTraceDao.java
@@ -28,11 +28,11 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Repository;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 /**
  * @author intr3p1d
@@ -47,7 +47,7 @@ public class PinotExceptionTraceDao implements ExceptionTraceDao {
 
     private final String topic;
 
-    private final ListenableFutureCallback<SendResult<String, ExceptionMetaDataEntity>> resultCallback
+    private final BiConsumer<SendResult<String, ExceptionMetaDataEntity>, Throwable> resultCallback
             = KafkaCallbacks.loggingCallback("Kafka(ExceptionMetaDataEntity)", logger);
 
 
@@ -68,10 +68,10 @@ public void insert(List<ExceptionMetaData> exceptionMetaData) {
 
         for (ExceptionMetaData e : exceptionMetaData) {
             ExceptionMetaDataEntity dataEntity = mapper.toEntity(e);
-            ListenableFuture<SendResult<String, ExceptionMetaDataEntity>> response = this.kafkaExceptionMetaDataTemplate.send(
+            CompletableFuture<SendResult<String, ExceptionMetaDataEntity>> response = this.kafkaExceptionMetaDataTemplate.send(
                     topic, dataEntity
             );
-            response.addCallback(resultCallback);
+            response.whenComplete(resultCallback);
         }
     }
 }
diff --git a/log/log-collector/pom.xml b/log/log-collector/pom.xml
@@ -74,7 +74,7 @@
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka</artifactId>
-            <version>2.9.4</version>
+            <version>${spring.kafka.version}</version>
         </dependency>
 
         <dependency>

diff --git a/...ic/src/main/java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotMetricTagDao.java b/...ic/src/main/java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotMetricTagDao.java
@@ -30,11 +30,11 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Repository;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 /**
  * @author minwoo.jung
@@ -50,7 +50,7 @@ public class PinotMetricTagDao implements MetricTagDao {
     private final TagListTypeHandler tagListTypeHandler = new TagListTypeHandler();
     private final String topic;
 
-    private final ListenableFutureCallback<SendResult<String, MetricJsonTag>> resultCallback
+    private final BiConsumer<SendResult<String, MetricJsonTag>, Throwable> resultCallback
             = KafkaCallbacks.loggingCallback("Kafka(MetricJsonTag)", logger);
 
     public PinotMetricTagDao(SqlSessionTemplate sqlPinotSessionTemplate,
@@ -64,8 +64,8 @@ public PinotMetricTagDao(SqlSessionTemplate sqlPinotSessionTemplate,
     @Override
     public void insertMetricTag(MetricTag metricTag) {
         MetricJsonTag metricJsonTag = MetricJsonTag.covertMetricJsonTag(tagListTypeHandler, metricTag);
-        ListenableFuture<SendResult<String, MetricJsonTag>> callBack = kafkaTagTemplate.send(topic, metricTag.getHostGroupName(), metricJsonTag);
-        callBack.addCallback(resultCallback);
+        CompletableFuture<SendResult<String, MetricJsonTag>> callBack = kafkaTagTemplate.send(topic, metricTag.getHostGroupName(), metricJsonTag);
+        callBack.whenComplete(resultCallback);
     }
 
     public static class MetricJsonTag {

diff --git a/.../java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotSystemMetricDataTypeDao.java b/.../java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotSystemMetricDataTypeDao.java
@@ -27,11 +27,11 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Repository;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 /**
  * @author minwoo.jung
@@ -47,7 +47,7 @@ public class PinotSystemMetricDataTypeDao implements SystemMetricDataTypeDao {
     private final KafkaTemplate<String, MetricData> kafkaDataTypeTemplate;
     private final String topic;
 
-    private final ListenableFutureCallback<SendResult<String, MetricData>> resultCallback
+    private final BiConsumer<SendResult<String, MetricData>, Throwable> resultCallback
             = KafkaCallbacks.loggingCallback("Kafka(MetricData)", logger);
 
     public PinotSystemMetricDataTypeDao(SqlSessionTemplate sqlPinotSessionTemplate,
@@ -71,7 +71,7 @@ public MetricData selectMetricDataType(MetricDataName metricDataName) {
 
     @Override
     public void updateMetricDataType(MetricData metricData) {
-        ListenableFuture<SendResult<String, MetricData>> callback = kafkaDataTypeTemplate.send(topic, metricData.getMetricName(), metricData);
-        callback.addCallback(resultCallback);
+        CompletableFuture<SendResult<String, MetricData>> callback = kafkaDataTypeTemplate.send(topic, metricData.getMetricName(), metricData);
+        callback.whenComplete(resultCallback);
     }
 }
diff --git a/...in/java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotSystemMetricDoubleDao.java b/...in/java/com/navercorp/pinpoint/metric/collector/dao/pinot/PinotSystemMetricDoubleDao.java
@@ -26,11 +26,11 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Repository;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 /**
  * @author Hyunjoon Cho
@@ -44,7 +44,7 @@ public class PinotSystemMetricDoubleDao implements SystemMetricDao<DoubleMetric>
 
     private final String topic;
 
-    private final ListenableFutureCallback<SendResult<String, SystemMetricView>> resultCallback
+    private final BiConsumer<SendResult<String, SystemMetricView>, Throwable> resultCallback
             = KafkaCallbacks.loggingCallback("Kafka(SystemMetricView)", logger);
 
     public PinotSystemMetricDoubleDao(KafkaTemplate<String, SystemMetricView> kafkaDoubleTemplate,
@@ -63,8 +63,8 @@ public void insert(String tenantId, String hostGroupName, String hostName, List<
         for (DoubleMetric doubleMetric : systemMetrics) {
             String kafkaKey = generateKafkaKey(doubleMetric);
             SystemMetricView systemMetricView = new SystemMetricView(tenantId, hostGroupName, doubleMetric);
-            ListenableFuture<SendResult<String, SystemMetricView>> callback = this.kafkaDoubleTemplate.send(topic, kafkaKey, systemMetricView);
-            callback.addCallback(resultCallback);
+            CompletableFuture<SendResult<String, SystemMetricView>> callback = this.kafkaDoubleTemplate.send(topic, kafkaKey, systemMetricView);
+            callback.whenComplete(resultCallback);
         }
     }
 

diff --git a/pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/util/KafkaCallbacks.java b/pinot/pinot-kafka/src/main/java/com/navercorp/pinpoint/pinot/kafka/util/KafkaCallbacks.java
@@ -2,21 +2,21 @@
 
 import org.apache.logging.log4j.Logger;
 import org.springframework.kafka.support.SendResult;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
-public final class KafkaCallbacks {
+import java.util.function.BiConsumer;
 
-    public static <T> ListenableFutureCallback<SendResult<String, T>> loggingCallback(String name, Logger logger) {
-        return new ListenableFutureCallback<>() {
-            @Override
-            public void onFailure(Throwable ex) {
-                logger.warn("{} onFailure:{}", name, ex.getMessage(), ex);
-            }
+public final class KafkaCallbacks {
 
+    public static <T> BiConsumer<SendResult<String, T>, Throwable> loggingCallback(String name, Logger logger) {
+        return new BiConsumer<>() {
             @Override
-            public void onSuccess(SendResult<String, T> result) {
-                if (logger.isDebugEnabled()) {
-                    logger.debug("{} onSuccess:{}", name, result);
+            public void accept(SendResult<String, T> result, Throwable throwable) {
+                if (throwable != null) {
+                    logger.warn("{} onFailure:{}", name, throwable.getMessage(), throwable);
+                } else {
+                    if (logger.isDebugEnabled()) {
+                        logger.debug("{} onSuccess:{}", name, result);
+                    }
                 }
             }
         };

diff --git a/pom.xml b/pom.xml
@@ -181,7 +181,7 @@
 <!--        <spring.security.version>5.8.1</spring.security.version>-->
         <spring.security.version>6.1.4</spring.security.version>
         <spring.boot2.version>2.7.13</spring.boot2.version>
-        <spring.kafka.version>2.9.11</spring.kafka.version>
+        <spring.kafka.version>3.0.12</spring.kafka.version>
 
         <spring.version>${spring6.version}</spring.version>
         <spring.boot.version>${spring.boot3.version}</spring.boot.version>

diff --git a/...collector/src/main/java/com/navercorp/pinpoint/uristat/collector/dao/PinotUriStatDao.java b/...collector/src/main/java/com/navercorp/pinpoint/uristat/collector/dao/PinotUriStatDao.java
@@ -10,11 +10,11 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Repository;
-import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.BiConsumer;
 
 @Repository
 public class PinotUriStatDao implements UriStatDao {
@@ -24,7 +24,7 @@ public class PinotUriStatDao implements UriStatDao {
 
     private final String topic;
 
-    private final ListenableFutureCallback<SendResult<String, UriStat>> resultCallback
+    private final BiConsumer<SendResult<String, UriStat>, Throwable> resultCallback
             = KafkaCallbacks.loggingCallback("Kafka(UriStat)", logger);
 
     public PinotUriStatDao(@Qualifier("kafkaUriStatTemplate") KafkaTemplate<String, UriStat> kafkaUriStatTemplate,
@@ -38,8 +38,8 @@ public void insert(List<UriStat> data) {
         Objects.requireNonNull(data);
 
         for (UriStat uriStat : data) {
-            ListenableFuture<SendResult<String, UriStat>> response = this.kafkaUriStatTemplate.send(topic, uriStat.getApplicationName(), uriStat);
-            response.addCallback(resultCallback);
+            CompletableFuture<SendResult<String, UriStat>> response = this.kafkaUriStatTemplate.send(topic, uriStat.getApplicationName(), uriStat);
+            response.whenComplete(resultCallback);
         }
 
     }