From 8bc2e90ea44e74afae039a4ab80608d2e5143fac Mon Sep 17 00:00:00 2001
From: zren25 <zhen.ren.2@volvocars.com>
Date: Fri, 16 May 2025 09:12:36 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9kafka=E9=85=8D=E7=BD=AE?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../analytic/center/config/KafkaConfig.java   | 22 +++----------------
 1 file changed, 3 insertions(+), 19 deletions(-)

diff --git a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java
index 5e8f18c..2aeee8a 100644
--- a/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java
+++ b/ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/config/KafkaConfig.java
@@ -51,25 +51,7 @@ public class KafkaConfig {
 
         return new KafkaTemplate<>(factory);
     }
-/**
-    @Bean(name = "dccConsumerFactory")
-    public ConcurrentKafkaListenerContainerFactory<String, String> dccConsumerFactory(
-            @Value("${kafka.dcc.bootstrap-servers}") String bootstrapServers,
-            @Value("${kafka.dcc.consumer.group}") String groupId,
-            @Value("${kafka.dcc.consumer.key-deserializer}") String keyDeserializer,
-            @Value("${kafka.dcc.consumer.value-deserializer}") String valueDeserializer) {
-        Map<String, Object> props = new HashMap<>();
-        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
-        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, keyDeserializer);
-        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
-        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
-        ConcurrentKafkaListenerContainerFactory<String, String> factory =
-                new ConcurrentKafkaListenerContainerFactory<>();
-        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
-        return factory;
-    }
-*/
+
     @Bean(name = "analyticCenterConsumerFactory")
     public ConcurrentKafkaListenerContainerFactory<String, String> analyticCenterConsumerFactory(
             @Value("${analyticCenterKafka.bootstrap-servers}") String bootstrapServers,
@@ -83,6 +65,8 @@ public class KafkaConfig {
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, valueDeserializer);
         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000); // 会话1分钟
+        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 可选：允许更长的消费间隔
         ConcurrentKafkaListenerContainerFactory<String, String> factory =
                 new ConcurrentKafkaListenerContainerFactory<>();
         factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));