current not current+1

snowflakedb · Nov 1, 2023 · b254308 · b254308
1 parent 12da4ed
commit b254308
Showing 1 changed file with 3 additions and 3 deletions.
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java b/src/main/java/com/snowflake/kafka/connector/internal/streaming/TopicPartitionChannel.java
@@ -341,7 +341,7 @@ public void insertRecordToBuffer(SinkRecord kafkaSinkRecord) {
     // Accept the incoming record only if we don't have a valid offset token at server side, or the
     // incoming record offset is 1 + the processed offset
     if (currentProcessedOffset == NO_OFFSET_TOKEN_REGISTERED_IN_SNOWFLAKE
-        || kafkaSinkRecord.kafkaOffset() >= currentProcessedOffset + 1) {
+        || kafkaSinkRecord.kafkaOffset() >= currentProcessedOffset) {
       StreamingBuffer copiedStreamingBuffer = null;
       bufferLock.lock();
       try {
@@ -403,7 +403,7 @@ private boolean shouldIgnoreAddingRecordToBuffer(
     }
 
     // Don't ignore if we see the expected offset; otherwise log and skip
-    if ((kafkaSinkRecord.kafkaOffset() - currentProcessedOffset) == 1L) {
+    if (kafkaSinkRecord.kafkaOffset() == currentProcessedOffset) {
       LOGGER.debug(
           "Got the desired offset:{} from Kafka, we can add this offset to buffer for channel:{}",
           kafkaSinkRecord.kafkaOffset(),
@@ -957,7 +957,7 @@ private void resetChannelMetadataAfterRecovery(
       // Need to update the in memory processed offset otherwise if same offset is send again, it
       // might get rejected.
       this.offsetPersistedInSnowflake.set(offsetRecoveredFromSnowflake);
-      this.processedOffset.set(offsetRecoveredFromSnowflake);
+      this.processedOffset.set(offsetToResetInKafka);
 
       // State that there was some exception and only clear that state when we have received offset
       // starting from offsetRecoveredFromSnowflake