修改kafka验证

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/KafkaMessageScheduler.java

@@ -32,7 +32,6 @@ public class KafkaMessageScheduler {
             return false;
         }
         log.info("messageQueueSize: {}",messageQueue.size());
-        messageQueue.offer(task);
-        return true;
+        return messageQueue.offer(task);
     }
 }

ai-analytic-center-biz/src/main/java/com/volvo/ai/analytic/center/mq/NameplateKafkaConsumer.java

@@ -2,18 +2,14 @@
 package com.volvo.ai.analytic.center.mq;
 
 import com.alibaba.fastjson.JSON;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.volvo.ai.analytic.center.dto.corpus.NameplateTableKafkaDTO;
 import com.volvo.ai.analytic.center.service.TmNameplateCorpusService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.cloud.context.config.annotation.RefreshScope;
-import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.KafkaHeaders;
@@ -22,8 +18,6 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * @ClassName NameplateKafkaConsumer
@@ -42,16 +36,11 @@ public class NameplateKafkaConsumer {
 	@Autowired
 	private TmNameplateCorpusService tmNameplateCorpusService;
 
-	private final ObjectMapper objectMapper = new ObjectMapper();
+
 	@Resource
 	private RocketMQTemplate rocketMqTemplate;
 
-	@Autowired
-	@Qualifier("kafkaTaskExecutor")
-	public ExecutorService kafkaTaskExecutor;
 
-	@Autowired
-	public KafkaMessageScheduler kafkaMessageScheduler;
 	@KafkaListener(topics = "${analyticCenterKafka.consumer.topic}",
 			groupId = "${analyticCenterKafka.consumer.group}" ,
 			containerFactory = "analyticCenterConsumerFactory",
@@ -63,35 +52,30 @@ public class NameplateKafkaConsumer {
 		log.info("nameplateKafkaConsumer 当前线程: {}, 线程ID: {},计数：{}", Thread.currentThread().getName(), Thread.currentThread().getId());
 		log.info("nameplateKafkaConsumerMessage: {}", recordMessages);
 		// 初始化绑定 Consumer
-
-
 		if (StringUtils.isEmpty(recordMessages)) {
 			ack.acknowledge();
 			return;
 		}
-
-		NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
-		if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
-			ack.acknowledge();
-			return;
-		}
-
-
-		boolean isSubmit =  kafkaMessageScheduler.submit(() -> {
-			try {
-				// 真正的业务逻辑在这里执行
-				tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpusService.processItem(nameplate));
-
-			} catch (Exception e) {
-				log.error("消息处理失败", e);
+		try {
+			NameplateTableKafkaDTO tmNameplateCorpus = JSON.parseObject(recordMessages, NameplateTableKafkaDTO.class);
+			if (!"INSERT".equals(tmNameplateCorpus.getType()) || CollectionUtils.isEmpty(tmNameplateCorpus.getData())) {
+				ack.acknowledge();
+				return;
 			}
-		});
-		if (isSubmit){
-			log.info("nameplateKafkaConsumeracknowledge：{}");
-			ack.acknowledge(); // 手动提交 offset
-		}
-		log.info("nameplateKafkaConsumer消息处理完成，耗时：{}", System.currentTimeMillis() - startTime);
 
+			tmNameplateCorpus.getData().forEach(nameplate -> tmNameplateCorpusService.processItem(nameplate));
+			log.info("nameplateKafkaConsumeracknowledge：{}",partitionId, offset);
+			 // 手动提交 offset
+		} catch (Exception e) {
+			log.info("nameplateKafkaConsumerFailed to process message: {}",  e);
+		}finally {
+			try {
+				ack.acknowledge(); // 提交 offset
+			} catch (IllegalStateException e) {
+				log.warn("Offset 已提交，跳过重复提交");
+			}
+		}
+		log.info("nameplateKafkaConsumer消息处理开始，耗时：{}", System.currentTimeMillis() - startTime);
 	}
 
 }