commit

apply.sh

@@ -1,5 +1,5 @@
 
-version="1.8"
+version="1.9"
 docker build -t registry.cn-chengdu.aliyuncs.com/cxzx-spider/ali-kafka:$version .
 docker push registry.cn-chengdu.aliyuncs.com/cxzx-spider/ali-kafka:$version
 sed -i "s!{ { image } }!registry\.cn-chengdu\.aliyuncs\.com/cxzx-spider/ali-kafka:$version!g" *.yml

src/main/java/com/cxzx/Main.java

@@ -27,11 +27,9 @@ public class Main {
                 var kafkaDataList = new ArrayList<KafkaData>();
                 for (ConsumerRecord<String, String> record : consumer.poll(Duration.ofMillis(2000))) {
                     kafkaDataList.add(new KafkaData(record.offset(), record.value(), new Timestamp(record.timestamp())));
-                    var offsetMap = Map.of(new TopicPartition(record.topic(), record.partition()),
-                            new OffsetAndMetadata(record.offset() + 1));
-                    consumer.commitSync(offsetMap);
                 }
                 Tidb.insert(kafkaDataList);
+                consumer.commitSync();
             }
         } catch (Exception e) {
             log.error(e.getMessage(), e);