Merge pull request #352 from marklogic/feature/logging-tweak

Fixing logging for writing zip files

Author: rjrudin

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

@@ -249,7 +249,7 @@ private synchronized void writeFailedDocumentToArchive(DocumentWriteOperation fa
         } catch (Exception e) {
             ConnectorException ex = new ConnectorException(String.format(
                 "Unable to write failed documents to archive file at %s; URI of failed document: %s; cause: %s",
-                archiveWriter.getZipPath(), failedDoc.getUri(), e.getMessage()
+                archiveWriter.getZipFilePath(), failedDoc.getUri(), e.getMessage()
             ), e);
             this.writeFailure.compareAndSet(null, ex);
             throw ex;
@@ -266,7 +266,7 @@ private ZipFileWriter createArchiveWriter(SerializableConfiguration hadoopConfig
     private void closeArchiveWriter() {
         if (archiveWriter != null) {
             if (failedItemCount.get() > 0) {
-                Util.MAIN_LOGGER.info("Wrote failed documents to archive file at {}.", archiveWriter.getZipPath());
+                Util.MAIN_LOGGER.info("Wrote failed documents to archive file at {}.", archiveWriter.getZipFilePath());
             }
             archiveWriter.close();
         }

src/main/java/com/marklogic/spark/writer/file/DocumentFileBatch.java

@@ -40,24 +40,26 @@ public void commit(WriterCommitMessage[] messages) {
             int zipFileCount = 0;
             int zipEntryCount = 0;
             String path = null;
+            String zipFilePath = null;
             for (WriterCommitMessage message : messages) {
                 if (message instanceof FileCommitMessage) {
-                    path = ((FileCommitMessage)message).getPath();
+                    path = ((FileCommitMessage) message).getPath();
                     fileCount += ((FileCommitMessage) message).getFileCount();
                 } else if (message instanceof ZipCommitMessage) {
                     path = ((ZipCommitMessage)message).getPath();
+                    zipFilePath = ((ZipCommitMessage) message).getZipFilePath();
                     zipFileCount++;
                     zipEntryCount += ((ZipCommitMessage) message).getZipEntryCount();
                 }
             }
             if (fileCount == 1) {
-                Util.MAIN_LOGGER.info("Wrote 1 file to {}.", path);
+                Util.MAIN_LOGGER.info("Wrote 1 file to: {}.", path);
             } else if (fileCount > 1) {
-                Util.MAIN_LOGGER.info("Wrote {} files to {}.", fileCount, path);
+                Util.MAIN_LOGGER.info("Wrote {} files to: {}.", fileCount, path);
             } else if (zipFileCount == 1) {
-                Util.MAIN_LOGGER.info("Wrote 1 zip file containing {} entries to {}.", zipEntryCount, path);
+                Util.MAIN_LOGGER.info("Wrote 1 zip file containing {} entries to: {}.", zipEntryCount, zipFilePath);
             } else if (zipFileCount > 1) {
-                Util.MAIN_LOGGER.info("Wrote {} zip files containing a total of {} entries to {}.", zipFileCount, zipEntryCount, path);
+                Util.MAIN_LOGGER.info("Wrote {} zip files containing a total of {} entries to: {}.", zipFileCount, zipEntryCount, path);
             }
         }
     }

src/main/java/com/marklogic/spark/writer/file/ZipCommitMessage.java

@@ -8,17 +8,23 @@
 class ZipCommitMessage implements WriterCommitMessage {
 
     private final String path;
+    private final String zipFilePath;
     private final int zipEntryCount;
 
-    ZipCommitMessage(String path, int zipEntryCount) {
+    ZipCommitMessage(String path, String zipFilePath, int zipEntryCount) {
         this.path = path;
+        this.zipFilePath = zipFilePath;
         this.zipEntryCount = zipEntryCount;
     }
 
     String getPath() {
         return path;
     }
 
+    String getZipFilePath() {
+        return zipFilePath;
+    }
+
     int getZipEntryCount() {
         return zipEntryCount;
     }

src/main/java/com/marklogic/spark/writer/file/ZipFileWriter.java

@@ -31,7 +31,8 @@ public class ZipFileWriter implements DataWriter<InternalRow> {
     private final ContextSupport context;
     private final SerializableConfiguration hadoopConfiguration;
 
-    private final String zipPath;
+    private final String path;
+    private final String zipFilePath;
 
     // These can be instantiated lazily depending on which constructor is used.
     private ContentWriter contentWriter;
@@ -45,7 +46,8 @@ public class ZipFileWriter implements DataWriter<InternalRow> {
 
     public ZipFileWriter(String path, Map<String, String> properties, SerializableConfiguration hadoopConfiguration,
                          int partitionId, boolean createZipFileImmediately) {
-        this.zipPath = makeFilePath(path, partitionId);
+        this.path = path;
+        this.zipFilePath = makeFilePath(path, partitionId);
         this.context = new ContextSupport(properties);
         this.hadoopConfiguration = hadoopConfiguration;
         if (createZipFileImmediately) {
@@ -75,7 +77,7 @@ public void close() {
 
     @Override
     public WriterCommitMessage commit() {
-        return new ZipCommitMessage(zipPath, zipEntryCounter);
+        return new ZipCommitMessage(path, zipFilePath, zipEntryCounter);
     }
 
     @Override
@@ -84,7 +86,7 @@ public void abort() {
     }
 
     private void createZipFileAndContentWriter() {
-        Path filePath = new Path(zipPath);
+        Path filePath = new Path(zipFilePath);
         if (logger.isDebugEnabled()) {
             logger.debug("Will write to: {}", filePath);
         }
@@ -131,7 +133,7 @@ private String makeFilePath(String path, int partitionId) {
         return String.format("%s%s%s-%d.zip", path, File.separator, timestamp, partitionId);
     }
 
-    public String getZipPath() {
-        return zipPath;
+    public String getZipFilePath() {
+        return zipFilePath;
     }
 }