Merge pull request #173 from marklogic/feature/logging-improvement

Added some logging for writing documents

Author: rjrudin

src/main/java/com/marklogic/spark/reader/document/DocumentBatch.java

@@ -47,7 +47,7 @@ class DocumentBatch implements Batch {
         this.partitions = planner.makePartitions(estimate, serverTimestamp, forests);
 
         if (Util.MAIN_LOGGER.isInfoEnabled()) {
-            logger.info("Created {} partitions; query estimate: {}", partitions.length, estimate);
+            Util.MAIN_LOGGER.info("Created {} partitions; query estimate: {}", partitions.length, estimate);
         }
     }
 

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

@@ -60,7 +60,10 @@ public boolean useCommitCoordinator() {
 
     @Override
     public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
-        Util.MAIN_LOGGER.info("Number of partitions: {}", info.numPartitions());
+        int partitions = info.numPartitions();
+        int threadCount = writeContext.getThreadCount();
+        Util.MAIN_LOGGER.info("Number of partitions: {}; thread count per partition: {}; total threads used for writing: {}",
+            partitions, threadCount, partitions * threadCount);
         return (DataWriterFactory) determineWriterFactory();
     }
 

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

@@ -61,11 +61,15 @@ public StructType getSchema() {
         return schema;
     }
 
+    int getThreadCount() {
+        return (int)getNumericOption(Options.WRITE_THREAD_COUNT, 4, 1);
+    }
+
     WriteBatcher newWriteBatcher(DataMovementManager dataMovementManager) {
         WriteBatcher writeBatcher = dataMovementManager
             .newWriteBatcher()
             .withBatchSize((int) getNumericOption(Options.WRITE_BATCH_SIZE, 100, 1))
-            .withThreadCount((int) getNumericOption(Options.WRITE_THREAD_COUNT, 4, 1))
+            .withThreadCount(getThreadCount())
             // WriteBatcherImpl has its own warn-level logging which is a bit verbose, including more than just the
             // message from the server. This is intended to always show up and be associated with our Spark connector
             // and also to be more brief, just capturing the main message from the server.

src/test/java/com/marklogic/spark/writer/WriteRowsTest.java

@@ -98,7 +98,7 @@ void testWithCustomConfig() {
     @Test
     void invalidThreadCount() {
         DataFrameWriter writer = newWriter().option(Options.WRITE_THREAD_COUNT, 0);
-        ConnectorException ex = assertThrowsConnectorException(() -> writer.save());
+        ConnectorException ex = assertThrows(ConnectorException.class, () -> writer.save());
         assertEquals("Value of 'spark.marklogic.write.threadCount' option must be 1 or greater.", ex.getMessage());
         verifyNoDocsWereWritten();
     }