* Add more log traces for troubleshooting issues

Azure · Sep 18, 2024 · ca93d95 · ca93d95
1 parent 8013b9a
commit ca93d95
Showing 1 changed file with 8 additions and 7 deletions.
diff --git a/src/main/java/com/microsoft/azure/kusto/kafka/connect/sink/TopicPartitionWriter.java b/src/main/java/com/microsoft/azure/kusto/kafka/connect/sink/TopicPartitionWriter.java
@@ -167,20 +167,20 @@ private boolean hasStreamingSucceeded(@NotNull IngestionStatus status) {
     }
 
     private void backOffForRemainingAttempts(int retryAttempts, Exception exception, @NotNull SourceFile fileDescriptor) {
-        String logMessage = String.format("Writing {%s} failed records to miscellaneous dead-letter queue topic={%s}",
-                fileDescriptor.records.size(), dlqTopicName);
-
         if (retryAttempts < maxRetryAttempts) {
             // RetryUtil can be deleted if exponential backOff is not required, currently using constant backOff.
             // long sleepTimeMs = RetryUtil.computeExponentialBackOffWithJitter(retryAttempts, TimeUnit.SECONDS.toMillis(5));
             long sleepTimeMs = retryBackOffTime;
-            log.error("Failed to ingest records into Kusto, backing off and retrying ingesting records " +
-                    "after {} milliseconds.", sleepTimeMs);
+            if(exception!=null) {
+                log.error("Failed to ingest records into Kusto, backing off and retrying ingesting records " +
+                        "after {} milliseconds.", sleepTimeMs, exception);
+            }
             try {
                 TimeUnit.MILLISECONDS.sleep(sleepTimeMs);
             } catch (InterruptedException interruptedErr) {
                 if (isDlqEnabled && behaviorOnError != BehaviorOnError.FAIL) {
-                    log.warn(logMessage);
+                    log.warn("InterruptedException:: Writing {} failed records to miscellaneous dead-letter queue topic={}",
+                            fileDescriptor.records.size(), dlqTopicName);
                     fileDescriptor.records.forEach(this::sendFailedRecordToDlq);
                 }
                 throw new ConnectException(String.format("Retrying ingesting records into KustoDB was interrupted " +
@@ -189,7 +189,8 @@ private void backOffForRemainingAttempts(int retryAttempts, Exception exception,
             }
         } else {
             if (isDlqEnabled && behaviorOnError != BehaviorOnError.FAIL) {
-                log.warn(logMessage);
+                log.warn(String.format("Retries exhausted, writing {%s} failed records to miscellaneous dead-letter queue topic={%s}",
+                        fileDescriptor.records.size(), dlqTopicName));
                 fileDescriptor.records.forEach(this::sendFailedRecordToDlq);
             }
             throw new ConnectException("Retry attempts exhausted, failed to ingest records into KustoDB.", exception);