From 23c858527a0d905e28888139e9ee73fa52ffcb96 Mon Sep 17 00:00:00 2001
From: Michal Bobowski <michal.bobowski@snowflake.com>
Date: Wed, 11 Dec 2024 13:00:06 +0100
Subject: [PATCH] One log to rule them all

---
 .../kafka/connector/internal/KCLogger.java    |  4 ++++
 .../internal/SnowflakeSinkServiceV1.java      | 21 ++++++++++++-------
 2 files changed, 17 insertions(+), 8 deletions(-)

diff --git a/src/main/java/com/snowflake/kafka/connector/internal/KCLogger.java b/src/main/java/com/snowflake/kafka/connector/internal/KCLogger.java
index 91a9494ab..4db31357a 100644
--- a/src/main/java/com/snowflake/kafka/connector/internal/KCLogger.java
+++ b/src/main/java/com/snowflake/kafka/connector/internal/KCLogger.java
@@ -95,6 +95,10 @@ public void error(String format, Object... vars) {
     }
   }
 
+  public boolean isDebugEnabled() {
+    return logger.isDebugEnabled();
+  }
+
   private String getFormattedLogMessage(String format, Object... vars) {
     if (prependMdcContext) {
       String connCtx = MDC.get(MDC_CONN_CTX_KEY);
diff --git a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java
index 63a0fb205..15a01c93b 100644
--- a/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java
+++ b/src/main/java/com/snowflake/kafka/connector/internal/SnowflakeSinkServiceV1.java
@@ -1102,14 +1102,19 @@ private void purge(List<String> files) {
     private void moveToTableStage(List<String> failedFiles) {
       if (!failedFiles.isEmpty()) {
         OffsetContinuityRanges offsets = searchForMissingOffsets(failedFiles);
-        LOGGER.info(
-            "Moving failed files for pipe: {} to tableStage failedFileCount: {}, continuousOffsets:"
-                + " {}, missingOffsets: {}",
-            pipeName,
-            failedFiles.size(),
-            offsets.getContinuousOffsets(),
-            offsets.getMissingOffsets());
-        LOGGER.debug("Moving failed files: {}", failedFiles);
+        String baseLog =
+            String.format(
+                "Moving failed files for pipe: %s to tableStage failedFileCount: %d,"
+                    + " continuousOffsets: %s, missingOffsets: %s",
+                pipeName,
+                failedFiles.size(),
+                offsets.getContinuousOffsets(),
+                offsets.getMissingOffsets());
+        if (LOGGER.isDebugEnabled()) {
+          LOGGER.info("{} - {}", baseLog, failedFiles);
+        } else {
+          LOGGER.info(baseLog);
+        }
         conn.moveToTableStage(tableName, stageName, failedFiles);
       }
     }