RenameDir and DeleteDir on blob to push number of blobs operated on last API call.

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

@@ -871,7 +871,7 @@ public boolean delete(final Path f, final boolean recursive) throws IOException
     statIncrement(CALL_DELETE);
     Path qualifiedPath = makeQualified(f);
     TracingContext tracingContext = new TracingContext(clientCorrelationId,
-            fileSystemId, FSOperationType.DELETE, tracingHeaderFormat,
+            fileSystemId, FSOperationType.DELETE, true, tracingHeaderFormat,
             listener);
 
     if (shouldRedirect(FSOperationType.DELETE, tracingContext)) {
@@ -926,18 +926,17 @@ public FileStatus[] listStatus(final Path f) throws IOException {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.LISTSTATUS, true, tracingHeaderFormat,
           listener);
-      FileStatus[] result = abfsStore.listStatus(qualifiedPath, tracingContext);
+      FileStatus[] result = getAbfsStore().listStatus(qualifiedPath, tracingContext);
       if (getAbfsStore().getAbfsConfiguration().getPrefixMode()
           == PrefixMode.BLOB) {
         FileStatus renamePendingFileStatus
-            = abfsStore.getRenamePendingFileStatus(result);
+            = getAbfsStore().getRenamePendingFileStatus(result);
         if (renamePendingFileStatus != null) {
           RenameAtomicityUtils renameAtomicityUtils =
-              new RenameAtomicityUtils(this,
-                  renamePendingFileStatus.getPath(),
-                  abfsStore.getRedoRenameInvocation(tracingContext));
+              getRenameAtomicityUtilsForRedo(renamePendingFileStatus.getPath(),
+                  tracingContext);
           renameAtomicityUtils.cleanup(renamePendingFileStatus.getPath());
-          result = abfsStore.listStatus(qualifiedPath, tracingContext);
+          result = getAbfsStore().listStatus(qualifiedPath, tracingContext);
         }
       }
       return result;
@@ -947,6 +946,13 @@ public FileStatus[] listStatus(final Path f) throws IOException {
     }
   }
 
+  RenameAtomicityUtils getRenameAtomicityUtilsForRedo(final Path renamePendingFileStatus,
+      final TracingContext tracingContext) throws IOException {
+    return new RenameAtomicityUtils(this,
+        renamePendingFileStatus,
+        getAbfsStore().getRedoRenameInvocation(tracingContext));
+  }
+
   /**
    * Increment of an Abfs statistic.
    *
@@ -1048,7 +1054,7 @@ public synchronized void close() throws IOException {
   @Override
   public FileStatus getFileStatus(final Path f) throws IOException {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
-          fileSystemId, FSOperationType.GET_FILESTATUS, tracingHeaderFormat,
+          fileSystemId, FSOperationType.GET_FILESTATUS, true, tracingHeaderFormat,
           listener);
       return getFileStatus(f, tracingContext);
   }
@@ -1070,17 +1076,16 @@ private FileStatus getFileStatus(final Path path,
          * Get File Status over Blob Endpoint will Have an additional call
          * to check if directory is implicit.
          */
-        fileStatus = abfsStore.getFileStatus(qualifiedPath,
+        fileStatus = getAbfsStore().getFileStatus(qualifiedPath,
             tracingContext, useBlobEndpoint);
         if (getAbfsStore().getPrefixMode() == PrefixMode.BLOB
                 && fileStatus != null && fileStatus.isDirectory()
-          && abfsStore.isAtomicRenameKey(fileStatus.getPath().toUri().getPath())
-          && abfsStore.getRenamePendingFileStatusInDirectory(fileStatus,
+          && getAbfsStore().isAtomicRenameKey(fileStatus.getPath().toUri().getPath())
+          && getAbfsStore().getRenamePendingFileStatusInDirectory(fileStatus,
               tracingContext)) {
-        RenameAtomicityUtils renameAtomicityUtils = new RenameAtomicityUtils(
-            this,
+        RenameAtomicityUtils renameAtomicityUtils = getRenameAtomicityUtilsForRedo(
             new Path(fileStatus.getPath().toUri().getPath() + SUFFIX),
-            abfsStore.getRedoRenameInvocation(tracingContext));
+            tracingContext);
         renameAtomicityUtils.cleanup(
             new Path(fileStatus.getPath().toUri().getPath() + SUFFIX));
         throw new AbfsRestOperationException(HttpURLConnection.HTTP_NOT_FOUND,

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

@@ -56,6 +56,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hadoop.classification.VisibleForTesting;
 
@@ -1580,6 +1581,7 @@ private void renameBlobDir(final Path source,
     final ExecutorService renameBlobExecutorService
         = Executors.newFixedThreadPool(
         getAbfsConfiguration().getBlobDirRenameMaxThread());
+    AtomicInteger renamedBlob = new AtomicInteger(0);
     while(!listBlobConsumer.isCompleted()) {
       blobList = listBlobConsumer.consume();
       if(blobList == null) {
@@ -1611,6 +1613,7 @@ private void renameBlobDir(final Path source,
                     blobProperty.getPath(), source),
                 blobLease,
                 tracingContext);
+            renamedBlob.incrementAndGet();
           } catch (AzureBlobFileSystemException e) {
             LOG.error(String.format("rename from %s to %s for blob %s failed",
                 source, destination, blobProperty.getPath()), e);
@@ -1634,11 +1637,13 @@ private void renameBlobDir(final Path source,
     }
     renameBlobExecutorService.shutdown();
 
+    tracingContext.setOperatedBlobCount(renamedBlob.get() + 1);
     renameBlob(
         source, createDestinationPathForBlobPartOfRenameSrcDir(destination,
             source, source),
         srcDirBlobLease,
         tracingContext);
+    tracingContext.setOperatedBlobCount(null);
   }
 
   private Boolean isCreateOperationOnBlobEndpoint() {
@@ -1757,7 +1762,6 @@ private void deleteBlobPath(final Path path,
     if (!path.isRoot()) {
       listSrcBuilder.append(FORWARD_SLASH);
     }
-
     String listSrc = listSrcBuilder.toString();
     BlobList blobList = client.getListBlobs(null, listSrc, null, null,
         tracingContext).getResult().getBlobList();
@@ -1868,6 +1872,7 @@ private void createParentDirectory(final Path path,
   private void deleteOnConsumedBlobs(final Path srcPath,
       final ListBlobConsumer consumer,
       final TracingContext tracingContext) throws AzureBlobFileSystemException {
+    AtomicInteger deletedBlobCount = new AtomicInteger(0);
     String srcPathStr = srcPath.toUri().getPath();
     ExecutorService deleteBlobExecutorService = Executors.newFixedThreadPool(
         getAbfsConfiguration().getBlobDirDeleteMaxThread());
@@ -1885,6 +1890,7 @@ private void deleteOnConsumedBlobs(final Path srcPath,
             try {
               client.deleteBlobPath(blobProperty.getPath(), null,
                   tracingContext);
+              deletedBlobCount.incrementAndGet();
             } catch (AzureBlobFileSystemException ex) {
               if (ex instanceof AbfsRestOperationException
                   && ((AbfsRestOperationException) ex).getStatusCode()
@@ -1912,6 +1918,8 @@ private void deleteOnConsumedBlobs(final Path srcPath,
     } finally {
       deleteBlobExecutorService.shutdown();
     }
+
+    tracingContext.setOperatedBlobCount(deletedBlobCount.get() + 1);
     if (!srcPath.isRoot()) {
       try {
         LOG.debug(String.format("Deleting Path %s", srcPathStr));
@@ -1926,6 +1934,7 @@ private void deleteOnConsumedBlobs(final Path srcPath,
             String.format("Path %s is an implicit directory", srcPathStr));
       }
     }
+    tracingContext.setOperatedBlobCount(null);
   }
 
   public FileStatus getFileStatus(Path path, TracingContext tracingContext, boolean useBlobEndpoint) throws IOException {

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

@@ -78,6 +78,8 @@ public class TracingContext {
   public static final int MAX_CLIENT_CORRELATION_ID_LENGTH = 72;
   public static final String CLIENT_CORRELATION_ID_PATTERN = "[a-zA-Z0-9-]*";
 
+  private Integer operatedBlobCount = null;
+
   /**
    * Initialize TracingContext
    * @param clientCorrelationID Provided over config by client
@@ -187,6 +189,9 @@ public void constructHeader(AbfsHttpOperation httpOperation, String previousFail
               + getPrimaryRequestIdForHeader(retryCount > 0) + ":" + streamID
               + ":" + opType + ":" + retryCount;
       header = addFailureReasons(header, previousFailure) + ":" + fallbackDFSAppend;
+      if (operatedBlobCount != null) {
+        header += (":" + operatedBlobCount);
+      }
       break;
     case TWO_ID_FORMAT:
       header = clientCorrelationID + ":" + clientRequestId;
@@ -241,4 +246,19 @@ public String getHeader() {
     return header;
   }
 
+  public String getPrimaryRequestId() {
+    return primaryRequestId;
+  }
+
+  public void setOperatedBlobCount(Integer count) {
+    operatedBlobCount = count;
+  }
+
+  public Integer getOperatedBlobCount() {
+    return operatedBlobCount;
+  }
+
+  public FSOperationType getOpType() {
+    return opType;
+  }
 }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemDelete.java

@@ -21,6 +21,7 @@
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
@@ -414,4 +415,68 @@ public void testDeleteCheckIfParentLMTChange() throws Exception {
     Long newLmt = fs.getFileStatus(new Path("/dir1")).getModificationTime();
     Assertions.assertThat(lmt).isEqualTo(newLmt);
   }
+
+  /**
+   * Test to assert that the CID in src marker delete contains the
+   * total number of blobs operated in the delete directory.
+   * Also, to assert that all operations in the delete-directory flow have same
+   * primaryId and opType.
+   */
+  @Test
+  public void testDeleteEmitDeletionCountInClientRequestId() throws Exception {
+    AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
+    Assume.assumeTrue(getPrefixMode(fs) == PrefixMode.BLOB);
+    String dirPathStr = "/testDir/dir1";
+    fs.mkdirs(new Path(dirPathStr));
+    ExecutorService executorService = Executors.newFixedThreadPool(5);
+    List<Future> futures = new ArrayList<>();
+    for (int i = 0; i < 10; i++) {
+      final int iter = i;
+      Future future = executorService.submit(() -> {
+        return fs.create(new Path("/testDir/dir1/file" + iter));
+      });
+      futures.add(future);
+    }
+
+    for (Future future : futures) {
+      future.get();
+    }
+    executorService.shutdown();
+
+
+    AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
+    Mockito.doReturn(store).when(fs).getAbfsStore();
+    AbfsClient client = Mockito.spy(store.getClient());
+    store.setClient(client);
+
+    final TracingHeaderValidator tracingHeaderValidator
+        = new TracingHeaderValidator(
+        fs.getAbfsStore().getAbfsConfiguration().getClientCorrelationId(),
+        fs.getFileSystemId(), FSOperationType.DELETE, true, 0);
+    fs.registerListener(tracingHeaderValidator);
+
+    Mockito.doAnswer(answer -> {
+          Mockito.doAnswer(deleteAnswer -> {
+                if (dirPathStr.equalsIgnoreCase(
+                    ((Path) deleteAnswer.getArgument(0)).toUri().getPath())) {
+                  tracingHeaderValidator.setOperatedBlobCount(11);
+                  Object result = deleteAnswer.callRealMethod();
+                  tracingHeaderValidator.setOperatedBlobCount(null);
+                  return result;
+                }
+                return deleteAnswer.callRealMethod();
+              })
+              .when(client)
+              .deleteBlobPath(Mockito.any(Path.class),
+                  Mockito.nullable(String.class),
+                  Mockito.any(TracingContext.class));
+
+          return answer.callRealMethod();
+        })
+        .when(store)
+        .delete(Mockito.any(Path.class), Mockito.anyBoolean(),
+            Mockito.any(TracingContext.class));
+
+    fs.delete(new Path(dirPathStr), true);
+  }
 }