commit

apply.sh

@@ -1,5 +1,5 @@
 
-version="1.7"
+version="1.8"
 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,7 +27,8 @@ 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()));
+                    var offsetMap = Map.of(new TopicPartition(record.topic(), record.partition()),
+                            new OffsetAndMetadata(record.offset() + 1));
                     consumer.commitSync(offsetMap);
                 }
                 Tidb.insert(kafkaDataList);