Remove atomic dir action in fileStatus and listStatus from store class

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

@@ -1195,8 +1195,6 @@ public FileStatus getFileStatus(final Path path,
 
       perfInfo.registerSuccess(true);
 
-      getClient().takeGetPathStatusAtomicRenameKeyAction(path, tracingContext);
-
       return new VersionedFileStatus(
               transformedOwner,
               transformedGroup,
@@ -1313,26 +1311,20 @@ public String listStatus(final Path path, final String startFrom,
           Path entryPath = new Path(File.separator + entry.name());
           entryPath = entryPath.makeQualified(this.uri, entryPath);
 
-          final boolean actionTakenOnRenamePendingJson
-              = getClient().takeListPathAtomicRenameKeyAction(entryPath,
-              (int) contentLength,
-              tracingContext);
-          if (!actionTakenOnRenamePendingJson) {
-            fileStatuses.add(
-                new VersionedFileStatus(
-                    owner,
-                    group,
-                    fsPermission,
-                    hasAcl,
-                    contentLength,
-                    isDirectory,
-                    1,
-                    blockSize,
-                    lastModifiedMillis,
-                    entryPath,
-                    entry.eTag(),
-                    encryptionContext));
-          }
+          fileStatuses.add(
+              new VersionedFileStatus(
+                  owner,
+                  group,
+                  fsPermission,
+                  hasAcl,
+                  contentLength,
+                  isDirectory,
+                  1,
+                  blockSize,
+                  lastModifiedMillis,
+                  entryPath,
+                  entry.eTag(),
+                  encryptionContext));
         }
 
         perfInfo.registerSuccess(true);

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

@@ -55,6 +55,7 @@
 import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
 import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore;
 import org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
 import org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations;
 import org.apache.hadoop.fs.azurebfs.constants.HttpQueryParams;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsInvalidChecksumException;
@@ -602,6 +603,7 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur
         requestHeaders);
 
     op.execute(tracingContext);
+    fixAtomicEntriesInListResults(op, tracingContext);
     if (isEmptyListResults(op.getResult()) && is404CheckRequired) {
       // If the list operation returns no paths, we need to check if the path is a file.
       // If it is a file, we need to return the file in the list.
@@ -623,6 +625,34 @@ public AbfsRestOperation listPath(final String relativePath, final boolean recur
     return op;
   }
 
+  private void fixAtomicEntriesInListResults(final AbfsRestOperation op,
+      final TracingContext tracingContext) throws AzureBlobFileSystemException {
+    /*
+     * Crashed HBase log rename recovery is done by Filesystem.getFileStatus and
+     * Filesystem.listStatus.
+     */
+    if (tracingContext == null
+        || tracingContext.getOpType() != FSOperationType.LISTSTATUS
+        || op == null || op.getResult() == null
+        || op.getResult().getStatusCode() != HTTP_OK) {
+      return;
+    }
+    BlobListResultSchema listResultSchema
+        = (BlobListResultSchema) op.getResult().getListResultSchema();
+    if (listResultSchema == null) {
+      return;
+    }
+    List<BlobListResultEntrySchema> filteredEntries = new ArrayList<>();
+    for (BlobListResultEntrySchema entry : listResultSchema.paths()) {
+      if (!takeListPathAtomicRenameKeyAction(entry.path(),
+          (int) (long) entry.contentLength(), tracingContext)) {
+        filteredEntries.add(entry);
+      }
+    }
+
+    listResultSchema.withPaths(filteredEntries);
+  }
+
   private boolean isEmptyListResults(AbfsHttpOperation result) {
     return result != null && result.getStatusCode() == HTTP_OK &&
         result.getListResultSchema() != null &&
@@ -1107,7 +1137,18 @@ public AbfsRestOperation getPathStatus(final String path,
       final TracingContext tracingContext,
       final ContextEncryptionAdapter contextEncryptionAdapter)
       throws AzureBlobFileSystemException {
-    return this.getPathStatus(path, tracingContext, contextEncryptionAdapter, true);
+    AbfsRestOperation op = this.getPathStatus(path, tracingContext,
+        contextEncryptionAdapter, true);
+    /*
+     * Crashed HBase log-folder rename can be recovered by FileSystem#getFileStatus
+     * and FileSystem#listStatus calls.
+     */
+    if (tracingContext != null
+        && tracingContext.getOpType() == FSOperationType.GET_FILESTATUS
+        && op.getResult() != null && checkIsDir(op.getResult())) {
+      takeGetPathStatusAtomicRenameKeyAction(new Path(path), tracingContext);
+    }
+    return op;
   }
 
   /**
@@ -1440,9 +1481,16 @@ public boolean isAtomicRenameKey(String key) {
     return isKeyForDirectorySet(key, azureAtomicRenameDirSet);
   }
 
-  @Override
+  /**
+   * Action to be taken when atomic-key is present on a getPathStatus path.
+   *
+   * @param path path of the pendingJson for the atomic path.
+   * @param tracingContext tracing context.
+   *
+   * @throws AzureBlobFileSystemException server error or the path is renamePending json file and action is taken.
+   */
   public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
-      final TracingContext tracingContext) throws IOException {
+      final TracingContext tracingContext) throws AzureBlobFileSystemException {
     if (path == null || path.isRoot() || !isAtomicRenameKey(path.toUri().getPath())) {
       return;
     }
@@ -1494,10 +1542,19 @@ public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
     }
   }
 
-  @Override
+  /**
+   * Action to be taken when atomic-key is present on a listPath path.
+   *
+   * @param path path of the pendingJson for the atomic path.
+   * @param renamePendingJsonLen length of the pendingJson file.
+   * @param tracingContext tracing context.
+   *
+   * @return true if action is taken.
+   * @throws AzureBlobFileSystemException server error
+   */
   public boolean takeListPathAtomicRenameKeyAction(final Path path,
       final int renamePendingJsonLen, final TracingContext tracingContext)
-      throws IOException {
+      throws AzureBlobFileSystemException {
     if (path == null || path.isRoot() || !isAtomicRenameKey(
         path.toUri().getPath()) || !path.toUri()
         .getPath()

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

@@ -1222,31 +1222,6 @@ public boolean isMetricCollectionEnabled() {
     return isMetricCollectionEnabled;
   }
 
-  /**
-   * Action to be taken when atomic-key is present on a getPathStatus path.
-   *
-   * @param path path of the pendingJson for the atomic path.
-   * @param tracingContext tracing context.
-   *
-   * @throws IOException server error or the path is renamePending json file and action is taken.
-   */
-  public abstract void takeGetPathStatusAtomicRenameKeyAction(final Path path,
-      final TracingContext tracingContext) throws IOException;
-
-  /**
-   * Action to be taken when a pendingJson is child of an atomic-key listing.
-   *
-   * @param path path of the pendingJson for the atomic path.
-   * @param renamePendingJsonLen length of the json file
-   * @param tracingContext tracing context.
-   *
-   * @return if path is atomicRenameJson and action is taken.
-   *
-   * @throws IOException server error
-   */
-  public abstract boolean takeListPathAtomicRenameKeyAction(final Path path,
-      final int renamePendingJsonLen, final TracingContext tracingContext) throws IOException;
-
   class TimerTaskImpl extends TimerTask {
     TimerTaskImpl() {
       runningTimerTask = this;

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

@@ -1329,18 +1329,6 @@ public String decodeAttribute(byte[] value) throws UnsupportedEncodingException
     return new String(value, XMS_PROPERTIES_ENCODING_ASCII);
   }
 
-  @Override
-  public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
-      final TracingContext tracingContext) throws IOException {
-
-  }
-
-  @Override
-  public boolean takeListPathAtomicRenameKeyAction(final Path path,
-      final int renamePendingJsonLen, final TracingContext tracingContext) throws IOException {
-    return false;
-  }
-
   private String convertXmsPropertiesToCommaSeparatedString(final Map<String,
       String> properties) throws CharacterCodingException {
     StringBuilder commaSeparatedProperties = new StringBuilder();

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

@@ -105,7 +105,7 @@ int getMaxConsumptionParallelism() {
   /**
    * Orchestrates the rename operation.
    */
-  public AbfsClientRenameResult execute() throws IOException {
+  public AbfsClientRenameResult execute() throws AzureBlobFileSystemException {
     PathInformation pathInformation = new PathInformation();
     boolean result = false;
     if (preCheck(src, dst, pathInformation)) {
@@ -158,7 +158,7 @@ public AbfsClientRenameResult execute() throws IOException {
     }
   }
 
-  private boolean finalSrcRename() throws IOException {
+  private boolean finalSrcRename() throws AzureBlobFileSystemException {
     tracingContext.setOperatedBlobCount(operatedBlobCount.get() + 1);
     try {
       return renameInternal(src, dst);
@@ -168,8 +168,7 @@ private boolean finalSrcRename() throws IOException {
   }
 
   @VisibleForTesting
-  public RenameAtomicity getRenameAtomicity(final PathInformation pathInformation)
-      throws IOException {
+  public RenameAtomicity getRenameAtomicity(final PathInformation pathInformation) {
     return new RenameAtomicity(src,
         dst,
         new Path(src.getParent(), src.getName() + RenameAtomicity.SUFFIX),
@@ -515,8 +514,8 @@ private PathInformation getPathInformation(Path path,
       TracingContext tracingContext)
       throws AzureBlobFileSystemException {
     try {
-      AbfsRestOperation op = abfsClient.getPathStatus(path.toString(), false,
-          tracingContext, null);
+      AbfsRestOperation op = abfsClient.getPathStatus(path.toString(),
+          tracingContext, null, true);
 
       return new PathInformation(true,
           abfsClient.checkIsDir(op.getResult()),

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

@@ -33,6 +33,7 @@
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.AppendRequestParameters;
@@ -122,7 +123,7 @@ public RenameAtomicity(final Path renameJsonPath,
   /**
    * Redo the rename operation from the JSON file.
    */
-  public void redo() throws IOException {
+  public void redo() throws AzureBlobFileSystemException {
     byte[] buffer = readRenamePendingJson(renameJsonPath, renamePendingJsonLen);
     String contents = new String(buffer, Charset.defaultCharset());
     try {
@@ -203,15 +204,15 @@ void createRenamePendingJson(Path path, byte[] bytes)
    * @return Length of the JSON file.
    */
   @VisibleForTesting
-  public int preRename() throws IOException {
+  public int preRename() throws AzureBlobFileSystemException {
     String makeRenamePendingFileContents = makeRenamePendingFileContents(
         srcEtag);
 
     try {
       createRenamePendingJson(renameJsonPath,
           makeRenamePendingFileContents.getBytes(StandardCharsets.UTF_8));
       return makeRenamePendingFileContents.length();
-    } catch (IOException e) {
+    } catch (AzureBlobFileSystemException e) {
       /*
        * Scenario: file has been deleted by parallel thread before the RenameJSON
        * could be written and flushed. In such case, there has to be one retry of
@@ -246,7 +247,7 @@ private boolean isPreRenameRetriableException(IOException e) {
     return false;
   }
 
-  public void postRename() throws IOException {
+  public void postRename() throws AzureBlobFileSystemException {
     deleteRenamePendingJson();
   }
 
@@ -272,14 +273,17 @@ private void deleteRenamePendingJson() throws AzureBlobFileSystemException {
    * @return JSON string which represents the operation.
    */
   private String makeRenamePendingFileContents(String eTag) throws
-      JsonProcessingException {
+      AzureBlobFileSystemException {
 
     final RenamePendingJsonFormat renamePendingJsonFormat = new RenamePendingJsonFormat();
     renamePendingJsonFormat.setOldFolderName(src.toUri().getPath());
     renamePendingJsonFormat.setNewFolderName(dst.toUri().getPath());
     renamePendingJsonFormat.setETag(eTag);
-
-    return objectMapper.writeValueAsString(renamePendingJsonFormat);
+    try {
+      return objectMapper.writeValueAsString(renamePendingJsonFormat);
+    } catch (JsonProcessingException e) {
+      throw new AbfsDriverException(e);
+    }
   }
 
   /**

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

@@ -294,6 +294,10 @@ public String getPosition() {
     return position;
   }
 
+  public FSOperationType getOpType() {
+    return opType;
+  }
+
   /**
    * Sets the ingress handler.
    *