[accumulo] branch 2.0 updated: Make bulk import move idempotent (#1246)

[kt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

kturner pushed a commit to branch 2.0
in repository https://gitbox.apache.org/repos/asf/accumulo.git


The following commit(s) were added to refs/heads/2.0 by this push:
     new 7268fcf  Make bulk import move idempotent (#1246)
7268fcf is described below

commit 7268fcf9537165c6f8fbdd70f833ead1699de32c
Author: Keith Turner <kt...@apache.org>
AuthorDate: Wed Jul 3 09:34:59 2019 -0400

    Make bulk import move idempotent (#1246)
---
 .../master/tableOps/bulkVer2/BulkImportMove.java   | 25 +++++++++++++++++++++-
 1 file changed, 24 insertions(+), 1 deletion(-)

diff --git a/server/master/src/main/java/org/apache/accumulo/master/tableOps/bulkVer2/BulkImportMove.java b/server/master/src/main/java/org/apache/accumulo/master/tableOps/bulkVer2/BulkImportMove.java
index 9da76ab..83480fb 100644
--- a/server/master/src/main/java/org/apache/accumulo/master/tableOps/bulkVer2/BulkImportMove.java
+++ b/server/master/src/main/java/org/apache/accumulo/master/tableOps/bulkVer2/BulkImportMove.java
@@ -16,6 +16,7 @@
  */
 package org.apache.accumulo.master.tableOps.bulkVer2;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -109,7 +110,29 @@ class BulkImportMove extends MasterRepo {
       results.add(workers.submit(() -> {
         final Path originalPath = new Path(sourceDir, renameEntry.getKey());
         Path newPath = new Path(bulkDir, renameEntry.getValue());
-        boolean success = fs.rename(originalPath, newPath);
+        boolean success;
+        try {
+          success = fs.rename(originalPath, newPath);
+        } catch (IOException e) {
+          // The rename could have failed because this is the second time its running (failures
+          // could cause this to run multiple times).
+          if (!fs.exists(newPath) || fs.exists(originalPath)) {
+            throw e;
+          }
+
+          log.debug(
+              "Ingoring rename exception because destination already exists. tid: {} orig: {} new: {}",
+              fmtTid, originalPath, newPath, e);
+          success = true;
+        }
+
+        if (!success && fs.exists(newPath) && !fs.exists(originalPath)) {
+          log.debug(
+              "Ingoring rename failure because destination already exists. tid: {} orig: {} new: {}",
+              fmtTid, originalPath, newPath);
+          success = true;
+        }
+
         if (success && log.isTraceEnabled())
           log.trace("tid {} moved {} to {}", fmtTid, originalPath, newPath);
         return success;