HADOOP-19445: ABFS: [FnsOverBlob][Tests] Add Tests For Negative Scenarios Identified for Rename Operation

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

@@ -54,6 +54,7 @@
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.DIRECTORIES_CREATED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.DIRECTORIES_DELETED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.ERROR_IGNORED;
+import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_COPIED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_CREATED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_DELETED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.GET_RESPONSES;
@@ -134,7 +135,8 @@ public class AbfsCountersImpl implements AbfsCounters {
       SERVER_UNAVAILABLE,
       RENAME_RECOVERY,
       METADATA_INCOMPLETE_RENAME_FAILURES,
-      RENAME_PATH_ATTEMPTS
+      RENAME_PATH_ATTEMPTS,
+      FILES_COPIED
   };
 
   private static final AbfsStatistic[] DURATION_TRACKER_LIST = {

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

@@ -109,7 +109,9 @@ public enum AbfsStatistic {
       "Number of times rename operation failed due to metadata being "
           + "incomplete"),
   RENAME_PATH_ATTEMPTS("rename_path_attempts",
-      "Number of times we attempt to rename a path internally");
+      "Number of times we attempt to rename a path internally"),
+  FILES_COPIED("files_copied",
+      "Total number of files copied from the object store.");
 
   private String statName;
   private String statDescription;

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

@@ -993,8 +993,6 @@ public Void call() throws Exception {
             delete(fs.getPath(), fs.isDirectory());
             if (fs.isDirectory()) {
               statIncrement(DIRECTORIES_DELETED);
-            } else {
-              statIncrement(FILES_DELETED);
             }
             return null;
           }

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

@@ -56,7 +56,6 @@ public enum AzureServiceErrorCode {
   OTHER_SERVER_THROTTLING("ServerBusy", HttpURLConnection.HTTP_UNAVAILABLE,
           "The server is currently unable to receive requests. Please retry your request."),
   INVALID_QUERY_PARAMETER_VALUE("InvalidQueryParameterValue", HttpURLConnection.HTTP_BAD_REQUEST, null),
-  INVALID_RENAME_DESTINATION("InvalidRenameDestinationPath", HttpURLConnection.HTTP_BAD_REQUEST, null),
   AUTHORIZATION_PERMISSION_MISS_MATCH("AuthorizationPermissionMismatch", HttpURLConnection.HTTP_FORBIDDEN, null),
   ACCOUNT_REQUIRES_HTTPS("AccountRequiresHttps", HttpURLConnection.HTTP_BAD_REQUEST, null),
   MD5_MISMATCH("Md5Mismatch", HttpURLConnection.HTTP_BAD_REQUEST,

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

@@ -84,7 +84,6 @@
 import static java.net.HttpURLConnection.HTTP_OK;
 import static java.net.HttpURLConnection.HTTP_PRECON_FAILED;
 import static org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.extractEtagHeader;
-import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.CALL_GET_FILE_STATUS;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ACQUIRE_LEASE_ACTION;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.AND_MARK;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APPEND_BLOB_TYPE;
@@ -424,7 +423,7 @@ private void fixAtomicEntriesInListResults(final AbfsRestOperation op,
     }
     List<BlobListResultEntrySchema> filteredEntries = new ArrayList<>();
     for (BlobListResultEntrySchema entry : listResultSchema.paths()) {
-      if (!takeListPathAtomicRenameKeyAction(entry.path(),
+      if (!takeListPathAtomicRenameKeyAction(entry.path(), entry.isDirectory(),
           entry.contentLength().intValue(), tracingContext)) {
         filteredEntries.add(entry);
       }
@@ -444,15 +443,14 @@ public void createNonRecursivePreCheck(Path parentPath,
       }
       getPathStatus(parentPath.toUri().getPath(), false,
           tracingContext, null);
+      incrementAbfsGetPathStatus();
     } catch (AbfsRestOperationException ex) {
       if (ex.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
         throw new FileNotFoundException("Cannot create file "
             + parentPath.toUri().getPath()
             + " because parent folder does not exist.");
       }
       throw ex;
-    } finally {
-      getAbfsCounters().incrementCounter(CALL_GET_FILE_STATUS, 1);
     }
   }
 
@@ -510,6 +508,7 @@ protected AbfsRestOperation createMarkerAtPath(final String path,
       final String eTag,
       final ContextEncryptionAdapter contextEncryptionAdapter,
       final TracingContext tracingContext) throws AzureBlobFileSystemException {
+    incrementAbfsCreateFile();
     return createPathRestOp(path, false, false, false, eTag,
         contextEncryptionAdapter, tracingContext);
   }
@@ -807,7 +806,6 @@ public AbfsClientRenameResult renamePath(final String source,
     BlobRenameHandler blobRenameHandler = getBlobRenameHandler(source,
         destination, sourceEtag, isAtomicRenameKey(source), tracingContext
     );
-    incrementAbfsRenamePath();
     if (blobRenameHandler.execute()) {
       final AbfsUriQueryBuilder abfsUriQueryBuilder
           = createDefaultUriQueryBuilder();
@@ -1751,6 +1749,7 @@ public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
       pendingJsonFileStatus = getPathStatus(
           pendingJsonPath.toUri().getPath(), tracingContext,
           null, false);
+      incrementAbfsGetPathStatus();
       if (checkIsDir(pendingJsonFileStatus.getResult())) {
         return;
       }
@@ -1805,11 +1804,11 @@ public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
    * @throws AzureBlobFileSystemException server error
    */
   private boolean takeListPathAtomicRenameKeyAction(final Path path,
-      final int renamePendingJsonLen,
+      final boolean isDirectory, final int renamePendingJsonLen,
       final TracingContext tracingContext)
       throws AzureBlobFileSystemException {
     if (path == null || path.isRoot() || !isAtomicRenameKey(
-        path.toUri().getPath()) || !path.toUri()
+        path.toUri().getPath()) || isDirectory || !path.toUri()
         .getPath()
         .endsWith(RenameAtomicity.SUFFIX)) {
       return false;
@@ -1837,7 +1836,7 @@ private boolean takeListPathAtomicRenameKeyAction(final Path path,
   }
 
   @VisibleForTesting
-  RenameAtomicity getRedoRenameAtomicity(final Path renamePendingJsonPath,
+  public RenameAtomicity getRedoRenameAtomicity(final Path renamePendingJsonPath,
       int fileLen,
       final TracingContext tracingContext) {
     return new RenameAtomicity(renamePendingJsonPath,

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

@@ -91,6 +91,10 @@
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
 
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
+import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.CALL_GET_FILE_STATUS;
+import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_COPIED;
+import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_CREATED;
+import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.FILES_DELETED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.RENAME_PATH_ATTEMPTS;
 import static org.apache.hadoop.fs.azurebfs.AzureBlobFileSystemStore.extractEtagHeader;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.APN_VERSION;
@@ -693,6 +697,34 @@ protected void incrementAbfsRenamePath() {
     abfsCounters.incrementCounter(RENAME_PATH_ATTEMPTS, 1);
   }
 
+  /**
+   * Increments AbfsCounters for get path status by 1.
+   */
+  protected void incrementAbfsGetPathStatus() {
+    abfsCounters.incrementCounter(CALL_GET_FILE_STATUS, 1);
+  }
+
+  /**
+   * Increments AbfsCounters for Delete File by 1.
+   */
+  protected void incrementAbfsDeleteFile() {
+    abfsCounters.incrementCounter(FILES_DELETED, 1);
+  }
+
+  /**
+   * Increments AbfsCounters for Create File by 1.
+   */
+  protected void incrementAbfsCreateFile() {
+    abfsCounters.incrementCounter(FILES_CREATED, 1);
+  }
+
+  /**
+   * Increments AbfsCounters for Copy Files by 1.
+   */
+  protected void incrementAbfsCopyFile() {
+    abfsCounters.incrementCounter(FILES_COPIED, 1);
+  }
+
   /**
    * Check if the rename request failure is post a retry and if earlier rename
    * request might have succeeded at back-end.

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

@@ -1161,6 +1161,8 @@ public AbfsRestOperation deletePath(final String path,
       } else {
         return idempotencyOp;
       }
+    } finally {
+      incrementAbfsDeleteFile();
     }
 
     return op;

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

@@ -90,6 +90,7 @@ int getMaxConsumptionParallelism() {
   private boolean deleteInternal(final Path path)
       throws AzureBlobFileSystemException {
     getAbfsClient().deleteBlobPath(path, null, tracingContext);
+    getAbfsClient().incrementAbfsDeleteFile();
     deleteCount.incrementAndGet();
     return true;
   }

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

@@ -31,7 +31,6 @@
 
 import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathIOException;
 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.exceptions.TimeoutException;
@@ -121,6 +120,7 @@ int getMaxConsumptionParallelism() {
    * @throws AzureBlobFileSystemException if server call fails
    */
   public boolean execute() throws AzureBlobFileSystemException {
+    getAbfsClient().incrementAbfsRenamePath();
     PathInformation pathInformation = getPathInformation(src, tracingContext);
     boolean result = false;
     if (preCheck(src, dst, pathInformation)) {
@@ -257,36 +257,14 @@ private boolean containsColon(Path p) {
   private boolean preCheck(final Path src, final Path dst,
       final PathInformation pathInformation)
       throws AzureBlobFileSystemException {
-    validateDestinationPath(src, dst);
+    validateDestinationIsNotSubDir(src, dst);
     validateSourcePath(pathInformation);
     validateDestinationPathNotExist(src, dst, pathInformation);
     validateDestinationParentExist(src, dst, pathInformation);
 
     return true;
   }
 
-  /**
-   * Validate if the format of the destination path is correct and if the destination
-   * path is not a sub-directory of the source path.
-   *
-   * @param src source path
-   * @param dst destination path
-   *
-   * @throws AbfsRestOperationException if the destination path is invalid
-   */
-  private void validateDestinationPath(final Path src, final Path dst)
-      throws AbfsRestOperationException {
-    if (containsColon(dst)) {
-      throw new AbfsRestOperationException(
-          HttpURLConnection.HTTP_BAD_REQUEST,
-          AzureServiceErrorCode.INVALID_RENAME_DESTINATION.getErrorCode(), null,
-          new PathIOException(dst.toUri().getPath(),
-              "Destination path contains colon"));
-    }
-
-    validateDestinationIsNotSubDir(src, dst);
-  }
-
   /**
    * Validate if the destination path is not a sub-directory of the source path.
    *
@@ -439,7 +417,9 @@ private boolean renameInternal(final Path path,
     boolean operated = false;
     try {
       copyPath(path, destinationPathForBlobPartOfRenameSrcDir, leaseId);
+      getAbfsClient().incrementAbfsCopyFile();
       getAbfsClient().deleteBlobPath(path, leaseId, tracingContext);
+      getAbfsClient().incrementAbfsDeleteFile();
       operated = true;
     } finally {
       if (abfsLease != null) {

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

@@ -35,6 +35,7 @@
 import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
 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.exceptions.FileSystemOperationUnhandledException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidConfigurationValueException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.BlobListResultSchema;
 import org.apache.hadoop.fs.azurebfs.contracts.services.ListResultEntrySchema;
@@ -119,7 +120,14 @@ private boolean takeAction(List<Path> paths)
         LOG.debug("Thread interrupted while taking action on path: {}",
             path.toUri().getPath());
       } catch (ExecutionException e) {
-        executionException = (AzureBlobFileSystemException) e.getCause();
+        LOG.debug("Execution exception while taking action on path: {}",
+            path.toUri().getPath());
+        if (e.getCause() instanceof AzureBlobFileSystemException) {
+          executionException = (AzureBlobFileSystemException) e.getCause();
+        } else {
+          executionException =
+              new FileSystemOperationUnhandledException(executionException);
+        }
       }
     }
     if (executionException != null) {
@@ -261,7 +269,7 @@ protected String listAndEnqueue(final ListBlobQueue listBlobQueue,
   protected void addPaths(final List<Path> paths,
       final ListResultSchema retrievedSchema) {
     for (ListResultEntrySchema entry : retrievedSchema.paths()) {
-      Path entryPath = new Path(ROOT_PATH, entry.name());
+      Path entryPath = new Path(ROOT_PATH + entry.name());
       if (!entryPath.equals(this.path)) {
         paths.add(entryPath);
       }

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

@@ -267,6 +267,8 @@ public int preRename() throws AzureBlobFileSystemException {
         }
       }
       throw e;
+    } finally {
+      abfsClient.incrementAbfsCreateFile();
     }
   }
 
@@ -310,6 +312,8 @@ private void deleteRenamePendingJson() throws AzureBlobFileSystemException {
         return;
       }
       throw e;
+    } finally {
+      abfsClient.incrementAbfsDeleteFile();
     }
   }
 

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

@@ -248,8 +248,13 @@ private static String replacedUrl(String baseUrl, String oldString, String newSt
    */
   public static boolean isKeyForDirectorySet(String key, Set<String> dirSet) {
     for (String dir : dirSet) {
-      if (dir.isEmpty() || key.startsWith(
-          dir + AbfsHttpConstants.FORWARD_SLASH)) {
+      // Ensure the directory ends with a forward slash
+      if (StringUtils.isNotEmpty(dir)
+          && !dir.endsWith(AbfsHttpConstants.FORWARD_SLASH)) {
+        dir += AbfsHttpConstants.FORWARD_SLASH;
+      }
+      // Return true if the directory is empty or the key starts with the directory
+      if (dir.isEmpty() || key.startsWith(dir)) {
         return true;
       }