Merge pull request #301 from marklogic/feature/stream-log-progress

MLE-17095 Fixed streaming issue where progress wasn't logged

Author: rjrudin

CONTRIBUTING.md

@@ -1,8 +1,7 @@
 This guide covers how to develop and test this project. It assumes that you have cloned this repository to your local
 workstation.
 
-Due to the use of the Sonar plugin for Gradle, you must use Java 11 or higher for developing and testing the project. 
-The `build.gradle` file for this project ensures that the connector is built to run on Java 8 or higher. 
+You must use Java 11 or higher for developing, testing, and building this project. 
 
 # Setup
 

src/main/java/com/marklogic/spark/writer/WriteBatcherDataWriter.java

@@ -260,6 +260,7 @@ private void writeDocumentViaPutOperation(DocumentWriteOperation writeOp) {
         final String uri = replaceSpacesInUriForPutEndpoint(writeOp.getUri());
         try {
             this.documentManager.write(uri, writeOp.getMetadata(), (GenericWriteHandle) writeOp.getContent());
+            writeContext.logBatchOnSuccess(1, 0);
             this.successItemCount.incrementAndGet();
         } catch (RuntimeException ex) {
             captureFailure(ex.getMessage(), uri);

src/main/java/com/marklogic/spark/writer/WriteContext.java

@@ -246,18 +246,22 @@ private void addRestTransformParams(ServerTransform transform, String paramsValu
     }
 
     private void logBatchOnSuccess(WriteBatch batch) {
-        int docCount = batch.getItems().length;
-        if (docCount > 0) {
+        int documentCount = batch.getItems().length;
+        if (documentCount > 0) {
             WriteEvent firstEvent = batch.getItems()[0];
             // If the first event is the item added by DMSDK for the default metadata object, ignore it when showing
             // the count of documents in the batch.
             if (firstEvent.getTargetUri() == null && firstEvent.getMetadata() != null) {
-                docCount--;
+                documentCount--;
             }
         }
-        WriteProgressLogger.logProgressIfNecessary(docCount);
-        if (logger.isTraceEnabled()) {
-            logger.trace("Wrote batch; length: {}; job batch number: {}", docCount, batch.getJobBatchNumber());
+        logBatchOnSuccess(documentCount, batch.getJobBatchNumber());
+    }
+
+    public void logBatchOnSuccess(int documentCount, long optionalJobBatchNumber) {
+        WriteProgressLogger.logProgressIfNecessary(documentCount);
+        if (logger.isTraceEnabled() && optionalJobBatchNumber > 0) {
+            logger.trace("Wrote batch; length: {}; job batch number: {}", documentCount, optionalJobBatchNumber);
         }
     }
 

src/test/java/com/marklogic/spark/reader/file/ReadGenericFilesStreamingTest.java

@@ -35,6 +35,7 @@ void stream() throws Exception {
 
         defaultWrite(dataset.write().format(CONNECTOR_IDENTIFIER)
             .option(Options.STREAM_FILES, true)
+            .option(Options.WRITE_LOG_PROGRESS, 1) // Included for manual verification of logging.
             .option(Options.WRITE_COLLECTIONS, "streamed-files")
             .option(Options.WRITE_URI_REPLACE, ".*/mixed-files,''"));