Conditional check added for log raw records

Author: inv-jishnu

data-loader/core/src/main/java/com/scalar/db/dataloader/core/dataimport/log/AbstractImportLogger.java

@@ -75,6 +75,10 @@ public void onTransactionBatchCompleted(ImportTransactionBatchResult batchResult
     if (shouldSkipLoggingSuccess(batchResult)) {
       return;
     }
+    // Skip logging records if log raw records is not enabled and execution is not success
+    if (!batchResult.isSuccess() && !config.isLogRawSourceRecordsEnabled()) {
+      return;
+    }
 
     logTransactionBatch(batchResult);
     notifyTransactionBatchCompleted(batchResult);

data-loader/core/src/test/java/com/scalar/db/dataloader/core/dataimport/log/SingleFileImportLoggerTest.java

@@ -25,6 +25,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -262,4 +263,28 @@ private void assertDataChunkStatusEquals(
     assertEquals(
         expected.getTotalDurationInMilliSeconds(), actual.getTotalDurationInMilliSeconds());
   }
+
+  @Test
+  void onTransactionBatchCompleted_ShouldNotLogOrNotify_WhenRawLoggingIsSkipped()
+      throws IOException {
+    // Arrange
+    ImportLoggerConfig config =
+        ImportLoggerConfig.builder()
+            .logDirectoryPath(tempDir.toString() + "/")
+            .isLogRawSourceRecordsEnabled(false)
+            .isLogSuccessRecordsEnabled(true) // success logging OFF
+            .build();
+
+    // Spy on logger to verify internal calls
+    SingleFileImportLogger loggerSpy =
+        Mockito.spy(new SingleFileImportLogger(config, logWriterFactory));
+
+    ImportTransactionBatchResult batch = createBatchResults(1, false).get(0); // success batch
+
+    // Act
+    loggerSpy.onTransactionBatchCompleted(batch);
+
+    // Assert
+    Mockito.verify(loggerSpy, Mockito.never()).logTransactionBatch(Mockito.any());
+  }
 }