Do atomic replace in transaction

civitaspo · Aug 22, 2017 · ecca877 · ecca877
1 parent 07ed7f7
commit ecca877
Showing 1 changed file with 18 additions and 10 deletions.
diff --git a/src/main/java/org/embulk/output/hdfs/HdfsFileOutputPlugin.java b/src/main/java/org/embulk/output/hdfs/HdfsFileOutputPlugin.java
@@ -124,9 +124,26 @@ public ConfigDiff transaction(ConfigSource config, int taskCount,
         configure(task);
 
         control.run(task.dump());
+
+        if (task.getAtomicMode()) {
+            atomicReplace(task);
+        }
+
         return Exec.newConfigDiff();
     }
 
+    private void atomicReplace(PluginTask task)
+    {
+        HdfsClient hdfsClient = HdfsClient.build(task);
+        String outputDir = getOutputSampleDir(task);
+        String safeWsWithOutput = Paths.get(task.getSafeWorkspace(), getOutputSampleDir(task)).toString();
+
+        if (!hdfsClient.swapDirectory(safeWsWithOutput, outputDir)) {
+            throw new DataException(String.format("Failed to swap: src: %s, dst: %s", safeWsWithOutput, outputDir));
+        }
+        logger.info("Swapped: src: {}, dst: {}", safeWsWithOutput, outputDir);
+    }
+
     @Override
     public ConfigDiff resume(TaskSource taskSource,
             int taskCount,
@@ -147,22 +164,13 @@ public void cleanup(TaskSource taskSource,
             List outputPaths = successTaskReport.get(List.class, "output_paths");
             for (Object path : outputPaths) {
                 if (task.getAtomicMode()) {
-                    logger.info("Move {} to {}", path, outputDir);
+                    logger.info("Created and Moved: {} to {}", path, outputDir);
                 }
                 else {
                     logger.info("Created: {}", path);
                 }
             }
         }
-
-        if (task.getAtomicMode()) {
-            HdfsClient hdfsClient = HdfsClient.build(task);
-            String safeWsWithOutput = Paths.get(task.getSafeWorkspace(), getOutputSampleDir(task)).toString();
-            if (!hdfsClient.swapDirectory(safeWsWithOutput, outputDir)) {
-                throw new DataException(String.format("Failed to swap: src: %s, dst: %s", safeWsWithOutput, outputDir));
-            }
-            logger.info("Swapped: src: {}, dst: {}", safeWsWithOutput, outputDir);
-        }
     }
 
     @Override