NOSNOW: improve troubleshooting experience for scenario with skipped offsets

src/main/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);
     }
   }
 

src/main/java/com/snowflake/kafka/connector/internal/streaming/DirectTopicPartitionChannel.java

@@ -324,13 +324,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);
     }
   }