admin/vlv-ai-analytic-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sonar

Browse Source
This commit is contained in:
zren25
2025-05-16 14:31:45 +08:00
parent 22dc5a8026
commit f7bd2c2708
1 changed files with 4 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -68,7 +68,7 @@ public class NameplateKafkaConsumer {
log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size()); log.info("nameplateKafkaConsumerParseType: {},size:{}", tmNameplateCorpus.getType(), tmNameplateCorpus.getData().size());
if(tmNameplateCorpus.getType().equals("INSERT")){ if(tmNameplateCorpus.getType().equals("INSERT")){
CountDownLatch latch = new CountDownLatch(tmNameplateCorpus.getData().size()); CountDownLatch latch = new CountDownLatch(tmNameplateCorpus.getData().size());
tmNameplateCorpus.getData().forEach(nameplate -> { tmNameplateCorpus.getData().forEach(nameplate ->
CompletableFuture.runAsync(() -> { CompletableFuture.runAsync(() -> {
try { try {
log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId()); log.info("nameplateKafkaConsumerCustomerFlowId: {}",nameplate.getCustomerFlowId());
@@ -80,15 +80,14 @@ public class NameplateKafkaConsumer {
log.info("nameplateKafkaConsumerack.acknowledge"); log.info("nameplateKafkaConsumerack.acknowledge");
latch.countDown(); latch.countDown();
} }
}, executor); }, executor));
});
boolean allDone = latch.await(2, TimeUnit.MINUTES);
log.info("nameplateKafkaConsumerack.acknowledge:{}",allDone);
try { try {
ack.acknowledge(); // 确保无论成功与否都提交 offset ack.acknowledge(); // 确保无论成功与否都提交 offset
} catch (IllegalStateException e) { } catch (IllegalStateException e) {
log.warn("Offset 已提交，跳过重复提交"); log.warn("Offset 已提交，跳过重复提交");
} }
boolean allDone = latch.await(2, TimeUnit.MINUTES);
log.info("nameplateKafkaConsumerack.acknowledge:{}",allDone);
} }
}else { }else {
ack.acknowledge(); // 空消息直接跳过 ack.acknowledge(); // 空消息直接跳过