revert tpchannel

snowflakedb · Nov 1, 2023 · 8cd02e8 · 8cd02e8
1 parent 65f6a05
commit 8cd02e8
Showing 1 changed file with 1 addition and 1 deletion.
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
@@ -403,7 +403,7 @@ private boolean shouldIgnoreAddingRecordToBuffer(
     }
 
     // Don't ignore if we see the expected offset; otherwise log and skip
-    if (kafkaSinkRecord.kafkaOffset() == (currentProcessedOffset - 1)) {
+    if ((kafkaSinkRecord.kafkaOffset() - currentProcessedOffset) == 1L) {
       LOGGER.debug(
           "Got the desired offset:{} from Kafka, we can add this offset to buffer for channel:{}",
           kafkaSinkRecord.kafkaOffset(),