fix

apache · Sep 4, 2024 · 65e229c · 65e229c
1 parent e8ce65b
commit 65e229c
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/.../main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java b/.../main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaRecordEmitter.java
@@ -68,6 +68,8 @@ public void emitRecord(
             } else {
                 deserializationSchema.deserialize(consumerRecord.value(), outputCollector);
             }
+            // consumerRecord.offset + 1 is the offset commit to Kafka and also the start offset
+            // for the next run
             splitState.setCurrentOffset(consumerRecord.offset() + 1);
         } catch (IOException e) {
             if (this.messageFormatErrorHandleWay == MessageFormatErrorHandleWay.SKIP) {