Rename Resume optimization: resume only if srcDir etag matches to the one with which rename was started.

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java

@@ -43,6 +43,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidConfigurationValueException;
 import org.apache.hadoop.fs.azurebfs.services.AbfsBlobLease;
 import org.apache.hadoop.fs.azurebfs.services.BlobProperty;
@@ -116,6 +117,7 @@
 import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_DEFAULT;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.*;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.BLOB_LEASE_ONE_MINUTE_DURATION;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.FORWARD_SLASH;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_BLOB_ENDPOINT;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DNS_PREFIX;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.WASB_DNS_PREFIX;
@@ -928,15 +930,35 @@ public FileStatus[] listStatus(final Path f) throws IOException {
           listener);
       FileStatus[] result = getAbfsStore().listStatus(qualifiedPath, tracingContext);
       if (getAbfsStore().getAbfsConfiguration().getPrefixMode()
-          == PrefixMode.BLOB) {
-        FileStatus renamePendingFileStatus
+          == PrefixMode.BLOB && getAbfsStore().isAtomicRenameKey(
+          qualifiedPath.toUri().getPath() + FORWARD_SLASH)) {
+        Pair<FileStatus, FileStatus> renamePendingJsonAndSrcFileStatusPair
             = getAbfsStore().getRenamePendingFileStatus(result);
-        if (renamePendingFileStatus != null) {
-          RenameAtomicityUtils renameAtomicityUtils =
-              getRenameAtomicityUtilsForRedo(renamePendingFileStatus.getPath(),
-                  tracingContext);
-          renameAtomicityUtils.cleanup(renamePendingFileStatus.getPath());
-          result = getAbfsStore().listStatus(qualifiedPath, tracingContext);
+        FileStatus renamePendingSrcFileStatus
+            = renamePendingJsonAndSrcFileStatusPair.getRight();
+        FileStatus renamePendingJsonFileStatus
+            = renamePendingJsonAndSrcFileStatusPair.getLeft();
+        if (renamePendingJsonFileStatus != null) {
+          final Boolean isRedone;
+          if (renamePendingSrcFileStatus != null) {
+            RenameAtomicityUtils renameAtomicityUtils =
+                getRenameAtomicityUtilsForRedo(
+                    renamePendingJsonFileStatus.getPath(),
+                    tracingContext,
+                    ((AzureBlobFileSystemStore.VersionedFileStatus) renamePendingSrcFileStatus).getEtag());
+            renameAtomicityUtils.cleanup(renamePendingJsonFileStatus.getPath());
+            isRedone = renameAtomicityUtils.isRedone();
+          } else {
+            isRedone = false;
+            getAbfsStore().delete(renamePendingJsonFileStatus.getPath(), true,
+                tracingContext);
+          }
+          if (isRedone) {
+            result = getAbfsStore().listStatus(qualifiedPath, tracingContext);
+          } else {
+            result = ArrayUtils.removeElement(result,
+                renamePendingJsonFileStatus);
+          }
         }
       }
       return result;
@@ -947,10 +969,10 @@ public FileStatus[] listStatus(final Path f) throws IOException {
   }
 
   RenameAtomicityUtils getRenameAtomicityUtilsForRedo(final Path renamePendingFileStatus,
-      final TracingContext tracingContext) throws IOException {
+      final TracingContext tracingContext, final String srcEtag) throws IOException {
     return new RenameAtomicityUtils(this,
         renamePendingFileStatus,
-        getAbfsStore().getRedoRenameInvocation(tracingContext));
+        getAbfsStore().getRedoRenameInvocation(tracingContext), srcEtag);
   }
 
   /**
@@ -1083,15 +1105,20 @@ private FileStatus getFileStatus(final Path path,
           && getAbfsStore().isAtomicRenameKey(fileStatus.getPath().toUri().getPath())
           && getAbfsStore().getRenamePendingFileStatusInDirectory(fileStatus,
               tracingContext)) {
-        RenameAtomicityUtils renameAtomicityUtils = getRenameAtomicityUtilsForRedo(
-            new Path(fileStatus.getPath().toUri().getPath() + SUFFIX),
-            tracingContext);
-        renameAtomicityUtils.cleanup(
-            new Path(fileStatus.getPath().toUri().getPath() + SUFFIX));
-        throw new AbfsRestOperationException(HttpURLConnection.HTTP_NOT_FOUND,
-            AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(), null,
-            new FileNotFoundException(
-                qualifiedPath + ": No such file or directory."));
+          RenameAtomicityUtils renameAtomicityUtils
+              = getRenameAtomicityUtilsForRedo(
+              new Path(fileStatus.getPath().toUri().getPath() + SUFFIX),
+              tracingContext,
+              ((AzureBlobFileSystemStore.VersionedFileStatus) fileStatus).getEtag());
+          renameAtomicityUtils.cleanup(
+              new Path(fileStatus.getPath().toUri().getPath() + SUFFIX));
+          if (renameAtomicityUtils.isRedone()) {
+            throw new AbfsRestOperationException(
+                HttpURLConnection.HTTP_NOT_FOUND,
+                AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(), null,
+                new FileNotFoundException(
+                    qualifiedPath + ": No such file or directory."));
+          }
       }
       return fileStatus;
     } catch (AzureBlobFileSystemException ex) {

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystemStore.java

@@ -58,6 +58,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.hadoop.classification.VisibleForTesting;
 
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidConfigurationValueException;
@@ -681,6 +682,7 @@ BlobProperty getBlobProperty(Path blobPath,
     blobProperty.setCopyStatus(opResult.getResponseHeader(X_MS_COPY_STATUS));
     blobProperty.setContentLength(
         Long.parseLong(opResult.getResponseHeader(CONTENT_LENGTH)));
+    blobProperty.setETag(extractEtagHeader(opResult));
     return blobProperty;
   }
 
@@ -1027,7 +1029,7 @@ private AbfsRestOperation createFileOrMarker(boolean isNormalBlob, String relati
   }
 
   // Fallback plan : default to v1 create flow which will hit dfs endpoint. Config to enable: "fs.azure.ingress.fallback.to.dfs".
-  public OutputStream createFile(final Path path, final FileSystem.Statistics statistics, final boolean overwrite,
+  public AbfsOutputStream createFile(final Path path, final FileSystem.Statistics statistics, final boolean overwrite,
       final FsPermission permission, final FsPermission umask,
       TracingContext tracingContext, HashMap<String, String> metadata) throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("createFile", "createPath")) {
@@ -1241,7 +1243,7 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
             .build();
   }
 
-  public void createDirectory(final Path path, final FileSystem.Statistics statistics, final FsPermission permission,
+  public String createDirectory(final Path path, final FileSystem.Statistics statistics, final FsPermission permission,
       final FsPermission umask,
       final Boolean checkParentChain,
       TracingContext tracingContext)
@@ -1256,13 +1258,14 @@ public void createDirectory(final Path path, final FileSystem.Statistics statist
         }
         boolean blobOverwrite = abfsConfiguration.isEnabledBlobMkdirOverwrite();
 
-        createDirectoryMarkerBlob(path, statistics, permission, umask, tracingContext,
+        AbfsOutputStream pathDirectoryOutputStream = createDirectoryMarkerBlob(
+            path, statistics, permission, umask, tracingContext,
             blobOverwrite);
         for (Path pathToCreate: keysToCreateAsFolder) {
           createDirectoryMarkerBlob(pathToCreate, statistics, permission, umask,
               tracingContext, blobOverwrite);
         }
-        return;
+        return pathDirectoryOutputStream.getETag();
       }
       boolean isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
       LOG.debug("Mkdir created via dfs endpoint for the given path {} and config value {} ",
@@ -1282,18 +1285,19 @@ public void createDirectory(final Path path, final FileSystem.Statistics statist
               isNamespaceEnabled ? getOctalNotation(umask) : null, false, null,
               tracingContext);
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
+      return op.getResult().getResponseHeader(HttpHeaderConfigurations.ETAG);
     }
   }
 
-  private void createDirectoryMarkerBlob(final Path path,
+  private AbfsOutputStream createDirectoryMarkerBlob(final Path path,
       final FileSystem.Statistics statistics,
       final FsPermission permission,
       final FsPermission umask,
       final TracingContext tracingContext,
       final boolean blobOverwrite) throws IOException {
     HashMap<String, String> metadata = new HashMap<>();
     metadata.put(X_MS_META_HDI_ISFOLDER, TRUE);
-    createFile(path, statistics, blobOverwrite,
+    return createFile(path, statistics, blobOverwrite,
         permission, umask, tracingContext, metadata);
   }
 
@@ -1632,20 +1636,22 @@ private void orchestrateBlobRenameDir(final Path source,
       blobPropOnSrcNullable = null;
     }
 
+    final String srcDirETag;
     if (blobPropOnSrcNullable == null) {
       /*
        * There is no marker-blob, the client has to create marker blob before
        * starting the rename.
        */
       LOG.debug("Source {} is a directory but there is no marker-blob",
           source);
-      createDirectory(source, null, FsPermission.getDirDefault(),
+      srcDirETag = createDirectory(source, null, FsPermission.getDirDefault(),
           FsPermission.getUMask(
               getAbfsConfiguration().getRawConfiguration()),
           true, tracingContext);
     } else {
       LOG.debug("Source {} is a directory but there is a marker-blob",
           source);
+      srcDirETag = blobPropOnSrcNullable.getETag();
     }
     /*
      * If source is a directory, all the blobs in the directory have to be
@@ -1661,7 +1667,7 @@ private void orchestrateBlobRenameDir(final Path source,
           BLOB_LEASE_ONE_MINUTE_DURATION,
           tracingContext);
       renameAtomicityUtils.preRename(
-          isCreateOperationOnBlobEndpoint());
+          isCreateOperationOnBlobEndpoint(), srcDirETag);
       isAtomicRename = true;
     } else {
       srcDirLease = null;
@@ -2828,8 +2834,21 @@ public void redo(final Path destination, final Path src)
         }
         String listSrc = listSrcBuilder.toString();
         getListBlobProducer(listSrc, listBlobQueue, null, tracingContext);
-        AbfsBlobLease abfsBlobLease = getBlobLease(src.toUri().getPath(),
-            BLOB_LEASE_ONE_MINUTE_DURATION, tracingContext);
+        final AbfsBlobLease abfsBlobLease;
+        try {
+           abfsBlobLease = getBlobLease(src.toUri().getPath(),
+              BLOB_LEASE_ONE_MINUTE_DURATION, tracingContext);
+        } catch (AbfsRestOperationException ex) {
+          /*
+           * The required blob might be deleted in between the last check (from
+           * GetFileStatus or ListStatus) and the leaseAcquire. Hence, catching
+           * HTTP_NOT_FOUND error.
+           */
+          if (ex.getStatusCode() == HTTP_NOT_FOUND) {
+            return;
+          }
+          throw ex;
+        }
         renameBlobDir(src, destination, tracingContext, listBlobQueue,
             abfsBlobLease, true);
       }
@@ -3051,17 +3070,28 @@ private AbfsPerfInfo startTracking(String callerName, String calleeName) {
   }
 
   /**
-   * Search for a FileStatus corresponding to a RenamePending JSON file.
-   * @param fileStatuses array of fileStatus from which JSON file has to be searched.
-   * @return filestatus corresponding to RenamePending JSON file.
+   * Returns a pair of fileStatuses for the renamePendingJSON file and the renameSource file.
+   * @param fileStatuses array of fileStatus from which pair has to be searched.
+   * @return Pair of FileStatus. Left of the pair is fileStatus of renamePendingJson file.
+   * Right of the pair is fileStatus of renameSource file.
    */
-  public FileStatus getRenamePendingFileStatus(final FileStatus[] fileStatuses) {
+  public Pair<FileStatus, FileStatus> getRenamePendingFileStatus(final FileStatus[] fileStatuses) {
+    Map<String, FileStatus> fileStatusMap = new HashMap<>();
+    FileStatus renamePendingJsonFileStatus = null;
+    String requiredRenameSrcPath = null;
     for (FileStatus fileStatus : fileStatuses) {
-      if (fileStatus.getPath().toUri().getPath().endsWith(SUFFIX)) {
-        return fileStatus;
+      String path = fileStatus.getPath().toUri().getPath();
+      if (path.equals(requiredRenameSrcPath)) {
+        return Pair.of(renamePendingJsonFileStatus, fileStatus);
+      }
+      fileStatusMap.put(path, fileStatus);
+      if (path.endsWith(SUFFIX)) {
+        renamePendingJsonFileStatus = fileStatus;
+        requiredRenameSrcPath = path.split(SUFFIX)[0];
       }
     }
-    return null;
+    return Pair.of(renamePendingJsonFileStatus,
+        fileStatusMap.get(requiredRenameSrcPath));
   }
 
   /**

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RenameAtomicityUtils.java

@@ -50,7 +50,7 @@
 /**
  * For a directory enabled for atomic-rename, before rename starts, a
  * file with -RenamePending.json suffix is created. In this file, the states required
- * for the rename are given. This file is created by {@link #preRename(Boolean)} ()} method.
+ * for the rename are given. This file is created by {@link #preRename(Boolean, String)} ()} method.
  * This is important in case the JVM process crashes during rename, the atomicity
  * will be maintained, when the job calls {@link AzureBlobFileSystem#listStatus(Path)}
  * or {@link AzureBlobFileSystem#getFileStatus(Path)}. On these API calls to filesystem,
@@ -66,6 +66,7 @@ public class RenameAtomicityUtils {
   private Path srcPath;
   private Path dstPath;
   private TracingContext tracingContext;
+  private Boolean isReDone;
 
   private static final int MAX_RENAME_PENDING_FILE_SIZE = 10000000;
   private static final int FORMATTING_BUFFER = 10000;
@@ -87,13 +88,18 @@ public RenameAtomicityUtils(final AzureBlobFileSystem azureBlobFileSystem,
   }
 
   public RenameAtomicityUtils(final AzureBlobFileSystem azureBlobFileSystem,
-      final Path path, final RedoRenameInvocation redoRenameInvocation)
+      final Path path, final RedoRenameInvocation redoRenameInvocation,
+      final String srcEtag)
       throws IOException {
     this.azureBlobFileSystem = azureBlobFileSystem;
     final RenamePendingFileInfo renamePendingFileInfo = readFile(path);
-    if (renamePendingFileInfo != null) {
+    if (renamePendingFileInfo != null
+        && renamePendingFileInfo.eTag.equalsIgnoreCase(srcEtag)) {
       redoRenameInvocation.redo(renamePendingFileInfo.destination,
           renamePendingFileInfo.src);
+      isReDone = true;
+    } else {
+      isReDone = false;
     }
   }
 
@@ -148,13 +154,17 @@ private RenamePendingFileInfo readFile(final Path redoFile)
     // initialize this object's fields
     JsonNode oldFolderName = json.get("OldFolderName");
     JsonNode newFolderName = json.get("NewFolderName");
+    JsonNode eTag = json.get("ETag");
+
     if (oldFolderName != null && StringUtils.isNotEmpty(
         oldFolderName.textValue())
         && newFolderName != null && StringUtils.isNotEmpty(
-        newFolderName.textValue())) {
+        newFolderName.textValue()) && eTag != null && StringUtils.isNotEmpty(
+        eTag.textValue())) {
       RenamePendingFileInfo renamePendingFileInfo = new RenamePendingFileInfo();
       renamePendingFileInfo.destination = new Path(newFolderName.textValue());
       renamePendingFileInfo.src = new Path(oldFolderName.textValue());
+      renamePendingFileInfo.eTag = eTag.textValue();
       return renamePendingFileInfo;
     }
     return null;
@@ -186,6 +196,7 @@ private void deleteRenamePendingFile(FileSystem fs, Path redoFile)
    *   OperationTime: "<YYYY-MM-DD HH:MM:SS.MMM>",
    *   OldFolderName: "<key>",
    *   NewFolderName: "<key>"
+   *   ETag: "<etag of the src-directory>"
    * }
    *
    * Here's a sample:
@@ -194,15 +205,17 @@ private void deleteRenamePendingFile(FileSystem fs, Path redoFile)
    *  OperationUTCTime: "2014-07-01 23:50:35.572",
    *  OldFolderName: "user/ehans/folderToRename",
    *  NewFolderName: "user/ehans/renamedFolder"
+   *  ETag: "ETag"
    * } }</pre>
    * @throws IOException Thrown when fail to write file.
    */
-  public void preRename(final Boolean isCreateOperationOnBlobEndpoint) throws IOException {
+  public void preRename(final Boolean isCreateOperationOnBlobEndpoint,
+      final String eTag) throws IOException {
     Path path = getRenamePendingFilePath();
     LOG.debug("Preparing to write atomic rename state to {}", path.toString());
     OutputStream output = null;
 
-    String contents = makeRenamePendingFileContents();
+    String contents = makeRenamePendingFileContents(eTag);
 
     // Write file.
     try {
@@ -261,18 +274,22 @@ private Throwable getWrappedException(final IOException e) {
    *
    * @return JSON string which represents the operation.
    */
-  private String makeRenamePendingFileContents() {
+  private String makeRenamePendingFileContents(String eTag) {
     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
     sdf.setTimeZone(TimeZone.getTimeZone("UTC"));
     String time = sdf.format(new Date());
+    if(!eTag.startsWith("\"") && !eTag.endsWith("\"")) {
+      eTag = quote(eTag);
+    }
 
     // Make file contents as a string. Again, quote file names, escaping
     // characters as appropriate.
     String contents = "{\n"
         + "  FormatVersion: \"1.0\",\n"
         + "  OperationUTCTime: \"" + time + "\",\n"
         + "  OldFolderName: " + quote(srcPath.toUri().getPath()) + ",\n"
-        + "  NewFolderName: " + quote(dstPath.toUri().getPath()) + "\n"
+        + "  NewFolderName: " + quote(dstPath.toUri().getPath()) + ",\n"
+        + "  ETag: " + eTag + "\n"
         + "}\n";
 
     return contents;
@@ -366,10 +383,15 @@ private Path getRenamePendingFilePath() {
   private static class RenamePendingFileInfo {
     public Path destination;
     public Path src;
+    public String eTag;
   }
 
   public static interface RedoRenameInvocation {
     void redo(Path destination, Path src) throws
         AzureBlobFileSystemException;
   }
+
+  public Boolean isRedone() {
+    return isReDone;
+  }
 }