NOSNOW: improve troubleshooting experience for scenario with skipped …

…offsets (snowflakedb#883) (cherry picked from commit 7172481)
confluentinc · Aug 22, 2024 · 0dc8ca0 · 0dc8ca0
1 parent 563fd8f
commit 0dc8ca0
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 12 deletions.
diff --git a/.../java/com/snowflake/kafka/connector/internal/streaming/BufferedTopicPartitionChannel.java b/.../java/com/snowflake/kafka/connector/internal/streaming/BufferedTopicPartitionChannel.java
@@ -402,13 +402,13 @@ public void insertRecord(SinkRecord kafkaSinkRecord, boolean isFirstRowPerPartit
  insertRecords(copiedStreamingBuffer);
  }
  } else {
- LOGGER.debug(
- "Skip adding offset:{} to buffer for channel:{} because"
- + " offsetPersistedInSnowflake:{}, processedOffset:{}",
- kafkaSinkRecord.kafkaOffset(),
+ LOGGER.warn(
+ "Channel {} - skipping current record - expected offset {} but received {}. The current"
+ + " offset stored in Snowflake: {}",
  this.getChannelNameFormatV1(),
- currentOffsetPersistedInSnowflake,
- currentProcessedOffset);
+ currentProcessedOffset,
+ kafkaSinkRecord.kafkaOffset(),
+ currentOffsetPersistedInSnowflake);
  }
  }
 

diff --git a/...in/java/com/snowflake/kafka/connector/internal/streaming/DirectTopicPartitionChannel.java b/...in/java/com/snowflake/kafka/connector/internal/streaming/DirectTopicPartitionChannel.java
@@ -335,13 +335,13 @@ public void insertRecord(SinkRecord kafkaSinkRecord, boolean isFirstRowPerPartit
  || kafkaSinkRecord.kafkaOffset() >= currentProcessedOffset + 1) {
  transformAndSend(kafkaSinkRecord);
  } else {
- LOGGER.debug(
- "Skip adding offset:{} to buffer for channel:{} because"
- + " offsetPersistedInSnowflake:{}, processedOffset:{}",
- kafkaSinkRecord.kafkaOffset(),
+ LOGGER.warn(
+ "Channel {} - skipping current record - expected offset {} but received {}. The current"
+ + " offset stored in Snowflake: {}",
  this.getChannelNameFormatV1(),
- currentOffsetPersistedInSnowflake,
- currentProcessedOffset);
+ currentProcessedOffset,
+ kafkaSinkRecord.kafkaOffset(),
+ currentOffsetPersistedInSnowflake);
  }
  }