Producer condition change + Rename Azcopy tests

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

@@ -409,6 +409,10 @@ public class AbfsConfiguration{
       FS_AZURE_PRODUCER_QUEUE_MAX_SIZE, DefaultValue = DEFAULT_FS_AZURE_PRODUCER_QUEUE_MAX_SIZE)
   private int producerQueueMaxSize;
 
+  @IntegerConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_CONSUMER_MAX_LAG, DefaultValue = DEFAULT_FS_AZURE_CONSUMER_MAX_LAG)
+  private int listingMaxConsumptionLag;
+
   @IntegerConfigurationValidatorAnnotation(ConfigurationKey =
       FS_AZURE_BLOB_DIR_RENAME_MAX_THREAD, DefaultValue = DEFAULT_FS_AZURE_BLOB_RENAME_THREAD)
   private int blobRenameDirConsumptionParallelism;
@@ -1446,6 +1450,10 @@ public int getProducerQueueMaxSize() {
     return producerQueueMaxSize;
   }
 
+  public int getListingMaxConsumptionLag() {
+    return listingMaxConsumptionLag;
+  }
+
   public int getBlobRenameDirConsumptionParallelism() {
     return blobRenameDirConsumptionParallelism;
   }

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

@@ -357,6 +357,12 @@ public static String accountProperty(String property, String account) {
       = "fs.azure.blob.atomic.rename.lease.refresh.duration";
   /**Maximum number of blob information enqueued in memory for rename or delete orchestration {@value}*/
   public static final String FS_AZURE_PRODUCER_QUEUE_MAX_SIZE = "fs.azure.blob.dir.list.producer.queue.max.size";
+  /**
+   * Maximum consumer lag (count of blob information which is yet to be taken for operation)
+   * in blob listing which can be tolerated before making producer to wait for
+   * consumer lag to become tolerable: {@value}.
+   */
+  public static final String FS_AZURE_CONSUMER_MAX_LAG = "fs.azure.blob.dir.list.consumer.max.lag";
   /**Maximum number of thread per blob-rename orchestration {@value}*/
   public static final String FS_AZURE_BLOB_DIR_RENAME_MAX_THREAD = "fs.azure.blob.dir.rename.max.thread";
   /**Maximum number of thread per blob-delete orchestration {@value}*/

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

@@ -170,10 +170,11 @@ public final class FileSystemConfigurations {
   public static final long DEFAULT_AZURE_BLOB_COPY_PROGRESS_WAIT_MILLIS = 1_000L;
   public static final long
       DEFAULT_AZURE_BLOB_ATOMIC_RENAME_LEASE_REFRESH_DURATION = 60_000L;
-  public static final int DEFAULT_FS_AZURE_PRODUCER_QUEUE_MAX_SIZE = 10000;
-
-  public static final int DEFAULT_FS_AZURE_BLOB_RENAME_THREAD = 5;
-  public static final int DEFAULT_FS_AZURE_BLOB_DELETE_THREAD = 5;
+  public static final int DEFAULT_FS_AZURE_PRODUCER_QUEUE_MAX_SIZE = 2 * DEFAULT_AZURE_LIST_MAX_RESULTS;
+  public static final int DEFAULT_FS_AZURE_CONSUMER_MAX_LAG = DEFAULT_AZURE_LIST_MAX_RESULTS;
+  public static final int DEFAULT_FS_AZURE_LISTING_ACTION_THREADS = 5;
+  public static final int DEFAULT_FS_AZURE_BLOB_RENAME_THREAD = DEFAULT_FS_AZURE_LISTING_ACTION_THREADS;
+  public static final int DEFAULT_FS_AZURE_BLOB_DELETE_THREAD = DEFAULT_FS_AZURE_LISTING_ACTION_THREADS;
   public static final int BLOCK_ID_LENGTH = 60;
   private FileSystemConfigurations() {}
 }

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

@@ -1489,7 +1489,7 @@ public boolean isAtomicRenameKey(String key) {
    *
    * @throws AzureBlobFileSystemException server error or the path is renamePending json file and action is taken.
    */
-  public void takeGetPathStatusAtomicRenameKeyAction(final Path path,
+  private void takeGetPathStatusAtomicRenameKeyAction(final Path path,
       final TracingContext tracingContext) throws AzureBlobFileSystemException {
     if (path == null || path.isRoot() || !isAtomicRenameKey(path.toUri().getPath())) {
       return;
@@ -1583,7 +1583,7 @@ public boolean takeListPathAtomicRenameKeyAction(final Path path,
   }
 
   @VisibleForTesting
-  RenameAtomicity getRedoRenameAtomicity(final Path path, int fileLen,
+  public RenameAtomicity getRedoRenameAtomicity(final Path path, int fileLen,
       final TracingContext tracingContext) {
     RenameAtomicity renameAtomicity = new RenameAtomicity(path,
         fileLen,

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

@@ -110,6 +110,13 @@ public AbfsClientRenameResult execute() throws AzureBlobFileSystemException {
     boolean result = false;
     if (preCheck(src, dst, pathInformation)) {
       RenameAtomicity renameAtomicity = null;
+      if (pathInformation.getIsDirectory()
+          && pathInformation.getIsImplicit()) {
+        AbfsRestOperation createMarkerOp = abfsClient.createPath(src.toUri().getPath(), false, false,
+            null,
+            false, null, null, tracingContext, false);
+        pathInformation.setETag(extractEtagHeader(createMarkerOp.getResult()));
+      }
       try {
         if (isAtomicRename) {
           /*
@@ -263,7 +270,8 @@ private void validateDestinationIsNotSubDir(final Path src,
   private void setSrcPathInformation(final Path src,
       final PathInformation pathInformation)
       throws AzureBlobFileSystemException {
-    pathInformation.copy(getPathInformation(src, tracingContext));
+    pathInformation.
+        copy(getPathInformation(src, tracingContext));
   }
 
   /**
@@ -519,12 +527,13 @@ private PathInformation getPathInformation(Path path,
 
       return new PathInformation(true,
           abfsClient.checkIsDir(op.getResult()),
-          extractEtagHeader(op.getResult()));
+          extractEtagHeader(op.getResult()),
+          op.getResult() instanceof AbfsHttpOperation.AbfsHttpOperationWithFixedResultForGetFileStatus);
     } catch (AzureBlobFileSystemException e) {
       if (e instanceof AbfsRestOperationException) {
         AbfsRestOperationException ex = (AbfsRestOperationException) e;
         if (ex.getStatusCode() == HttpURLConnection.HTTP_NOT_FOUND) {
-          return new PathInformation(false, false, null);
+          return new PathInformation(false, false, null, false);
         }
       }
       throw e;

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

@@ -30,16 +30,19 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.Path;
 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.InvalidConfigurationValueException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.BlobListResultSchema;
 import org.apache.hadoop.fs.azurebfs.contracts.services.ListResultEntrySchema;
 import org.apache.hadoop.fs.azurebfs.contracts.services.ListResultSchema;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_AZURE_LIST_MAX_RESULTS;
 
 /**
  * ListActionTaker is an abstract class that provides a way to list the paths
@@ -48,7 +51,8 @@
  */
 public abstract class ListActionTaker {
 
-  private static final Logger LOG = LoggerFactory.getLogger(ListActionTaker.class);
+  private static final Logger LOG = LoggerFactory.getLogger(
+      ListActionTaker.class);
 
   protected final Path path;
 
@@ -75,7 +79,8 @@ public ListActionTaker(Path path,
 
   abstract boolean takeAction(Path path) throws AzureBlobFileSystemException;
 
-  private boolean takeAction(List<Path> paths) throws AzureBlobFileSystemException {
+  private boolean takeAction(List<Path> paths)
+      throws AzureBlobFileSystemException {
     List<Future<Boolean>> futureList = new ArrayList<>();
     for (Path path : paths) {
       Future<Boolean> future = executorService.submit(() -> {
@@ -111,12 +116,12 @@ private boolean takeAction(List<Path> paths) throws AzureBlobFileSystemException
    * path and supply them to parallel thread for relevant action which is defined
    * in {@link #takeAction(Path)}.
    */
-  public boolean listRecursiveAndTakeAction() throws AzureBlobFileSystemException {
+  public boolean listRecursiveAndTakeAction()
+      throws AzureBlobFileSystemException {
     AbfsConfiguration configuration = abfsClient.getAbfsConfiguration();
     Thread producerThread = null;
     try {
-      ListBlobQueue listBlobQueue = new ListBlobQueue(
-          configuration.getProducerQueueMaxSize(), getMaxConsumptionParallelism());
+      ListBlobQueue listBlobQueue = createListBlobQueue(configuration);
       producerThread = new Thread(() -> {
         try {
           produceConsumableList(listBlobQueue);
@@ -150,43 +155,66 @@ public boolean listRecursiveAndTakeAction() throws AzureBlobFileSystemException
     }
   }
 
+  @VisibleForTesting
+  protected ListBlobQueue createListBlobQueue(final AbfsConfiguration configuration)
+      throws InvalidConfigurationValueException {
+    return new ListBlobQueue(
+        configuration.getProducerQueueMaxSize(), getMaxConsumptionParallelism(),
+        configuration.getListingMaxConsumptionLag());
+  }
+
   private void produceConsumableList(final ListBlobQueue listBlobQueue)
       throws AzureBlobFileSystemException {
     String continuationToken = null;
     do {
-      List<Path> paths = new ArrayList<>();
-      final int queueAvailableSize = listBlobQueue.availableSize();
-      if (queueAvailableSize == 0) {
-        break;
-      }
-      final AbfsRestOperation op;
-      try {
-         op = abfsClient.listPath(path.toUri().getPath(),
-            true,
-            queueAvailableSize, continuationToken,
-            tracingContext);
-      } catch (AzureBlobFileSystemException ex) {
-        throw ex;
-      } catch (IOException ex) {
-        throw new AbfsRestOperationException(-1, null,
-            "Unknown exception from listing: " + ex.getMessage(), ex);
-      }
-
-      ListResultSchema retrievedSchema = op.getResult().getListResultSchema();
-      if (retrievedSchema == null) {
-        continue;
-      }
-      continuationToken
-          = ((BlobListResultSchema) retrievedSchema).getNextMarker();
-      for (ListResultEntrySchema entry : retrievedSchema.paths()) {
-        Path entryPath = new Path(ROOT_PATH, entry.name());
-        if (!entryPath.equals(this.path)) {
-          paths.add(entryPath);
-        }
-      }
-      listBlobQueue.enqueue(paths);
+      continuationToken = listAndEnqueue(listBlobQueue, continuationToken);
     } while (!producerThreadToBeStopped.get() && continuationToken != null
         && !listBlobQueue.getConsumptionFailed());
     listBlobQueue.complete();
   }
+
+  @VisibleForTesting
+  protected String listAndEnqueue(final ListBlobQueue listBlobQueue,
+      String continuationToken) throws AzureBlobFileSystemException {
+    final int queueAvailableSizeForProduction = Math.min(
+        DEFAULT_AZURE_LIST_MAX_RESULTS,
+        listBlobQueue.availableSizeForProduction());
+    if (queueAvailableSizeForProduction == 0) {
+      return null;
+    }
+    final AbfsRestOperation op;
+    try {
+      op = abfsClient.listPath(path.toUri().getPath(),
+          true,
+          queueAvailableSizeForProduction, continuationToken,
+          tracingContext);
+    } catch (AzureBlobFileSystemException ex) {
+      throw ex;
+    } catch (IOException ex) {
+      throw new AbfsRestOperationException(-1, null,
+          "Unknown exception from listing: " + ex.getMessage(), ex);
+    }
+
+    ListResultSchema retrievedSchema = op.getResult().getListResultSchema();
+    if (retrievedSchema == null) {
+      return continuationToken;
+    }
+    continuationToken
+        = ((BlobListResultSchema) retrievedSchema).getNextMarker();
+    List<Path> paths = new ArrayList<>();
+    addPaths(paths, retrievedSchema);
+    listBlobQueue.enqueue(paths);
+    return continuationToken;
+  }
+
+  @VisibleForTesting
+  protected void addPaths(final List<Path> paths,
+      final ListResultSchema retrievedSchema) {
+    for (ListResultEntrySchema entry : retrievedSchema.paths()) {
+      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/ListBlobQueue.java

@@ -25,6 +25,9 @@
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AzureBlobFileSystemException;
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidConfigurationValueException;
+
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_CONSUMER_MAX_LAG;
 
 /**
  * Data-structure to hold the list of paths to be processed. The paths are
@@ -49,9 +52,40 @@ class ListBlobQueue {
 
   private volatile AzureBlobFileSystemException failureFromProducer;
 
-  ListBlobQueue(int maxSize, int consumeSetSize) {
+  /**
+   * Maximum number of entries in the queue allowed for letting the producer to
+   * produce. If the current size of the queue is greater than or equal to
+   * maxConsumptionLag, the producer will wait until the current size of the queue
+   * becomes lesser than maxConsumptionLag. This parameter is used to control the
+   * behavior of the producer-consumer pattern and preventing producer from
+   * rapidly producing very small amount of items.
+   * <p>
+   * For example, let's say maxSize is 10000 and maxConsumptionLag is 5000.
+   * The producer will stop producing when the current size of the queue is 5000
+   * and will wait until the current size of the queue becomes lesser than 5000.
+   * Once, the size becomes lesser than 5000, producer can produce (maxSize - currentSize)
+   * of items, which would make the current size of the queue to be 10000. Then again
+   * it will wait for 5000 items to be consumed before generating next 5000 items.
+   * <p>
+   * If this is not used, the producer will keep on producing items as soon as
+   * the queue become available with small size. Let say, 5 items got consumed,
+   * producer would make a server call for only 5 items and would populate the queue.
+   * <p>
+   * This mechanism would prevent producer making server calls for very small amount
+   * of items.
+   */
+  private final int maxConsumptionLag;
+
+  ListBlobQueue(int maxSize, int consumeSetSize, int maxConsumptionLag)
+      throws InvalidConfigurationValueException {
     this.maxSize = maxSize;
+    this.maxConsumptionLag = maxConsumptionLag;
     this.consumeSetSize = consumeSetSize;
+
+    if (maxConsumptionLag >= maxSize) {
+      throw new InvalidConfigurationValueException(FS_AZURE_CONSUMER_MAX_LAG,
+          "maxConsumptionLag should be lesser than maxSize");
+    }
   }
 
   void markProducerFailure(AzureBlobFileSystemException failure) {
@@ -102,27 +136,27 @@ private List<Path> dequeue() {
       pathListForConsumption.add(pathQueue.poll());
       counter++;
     }
-    if(counter > 0) {
+    if (counter > 0) {
       notify();
     }
     return pathListForConsumption;
   }
 
-  private synchronized int size() {
+  synchronized int size() {
     return pathQueue.size();
   }
 
   /**
-   * Returns the available size of the queue. This is calculated by subtracting the current size of the queue
-   * from its maximum size. If the queue is full, this method will wait until some elements are consumed and
-   * space becomes available. If consumption has failed, it immediately returns zero. This method is synchronized
-   * to prevent concurrent modifications of the queue.
+   * Returns the available size of the queue for production. This is calculated by subtracting
+   * the current size of the queue from its maximum size. This method waits until
+   * the current size of the queue becomes lesser than the maxConsumptionLag. This
+   * method is synchronized to prevent concurrent modifications of the queue.
    *
-   * @return the available size of the queue
+   * @return the available size of the queue.
    */
-  synchronized int availableSize() {
-    while(maxSize - size() <= 0) {
-      if(isConsumptionFailed) {
+  synchronized int availableSizeForProduction() {
+    while (size() >= maxConsumptionLag) {
+      if (isConsumptionFailed) {
         return 0;
       }
       try {

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

@@ -21,12 +21,14 @@
 public class PathInformation {
   private Boolean pathExists;
   private Boolean isDirectory;
+  private Boolean isImplicit;
   private String eTag;
 
-  public PathInformation(Boolean pathExists, Boolean isDirectory, String eTag) {
+  public PathInformation(Boolean pathExists, Boolean isDirectory, String eTag, Boolean isImplicit) {
     this.pathExists = pathExists;
     this.isDirectory = isDirectory;
     this.eTag = eTag;
+    this.isImplicit = isImplicit;
   }
 
   public PathInformation() {
@@ -36,6 +38,7 @@ public void copy(PathInformation pathInformation) {
     this.pathExists = pathInformation.getPathExists();
     this.isDirectory = pathInformation.getIsDirectory();
     this.eTag = pathInformation.getETag();
+    this.isImplicit = pathInformation.getIsImplicit();
   }
 
   public String getETag() {
@@ -49,4 +52,12 @@ public Boolean getPathExists() {
   public Boolean getIsDirectory() {
     return isDirectory;
   }
+
+  public Boolean getIsImplicit() {
+    return isImplicit;
+  }
+
+  void setETag(String eTag) {
+    this.eTag = eTag;
+  }
 }