Producer condition change + Rename Azcopy tests

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

@@ -77,7 +77,9 @@
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.*;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_LEASE_CREATE_NON_RECURSIVE;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.*;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_FS_AZURE_LEASE_CREATE_NON_RECURSIVE;
 
 /**
  * Configuration for Azure Blob FileSystem.
@@ -409,6 +411,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;
@@ -417,6 +423,10 @@ public class AbfsConfiguration{
       FS_AZURE_BLOB_DIR_DELETE_MAX_THREAD, DefaultValue = DEFAULT_FS_AZURE_BLOB_DELETE_THREAD)
   private int blobDeleteDirConsumptionParallelism;
 
+  @BooleanConfigurationValidatorAnnotation(ConfigurationKey =
+      FS_AZURE_LEASE_CREATE_NON_RECURSIVE, DefaultValue = DEFAULT_FS_AZURE_LEASE_CREATE_NON_RECURSIVE)
+  private boolean isLeaseOnCreateNonRecursiveEnabled;
+
   private String clientProvidedEncryptionKey;
   private String clientProvidedEncryptionKeySHA;
 
@@ -1446,11 +1456,19 @@ public int getProducerQueueMaxSize() {
     return producerQueueMaxSize;
   }
 
+  public int getListingMaxConsumptionLag() {
+    return listingMaxConsumptionLag;
+  }
+
   public int getBlobRenameDirConsumptionParallelism() {
     return blobRenameDirConsumptionParallelism;
   }
 
   public int getBlobDeleteDirConsumptionParallelism() {
     return blobDeleteDirConsumptionParallelism;
   }
+
+  public boolean isLeaseOnCreateNonRecursiveEnabled() {
+    return isLeaseOnCreateNonRecursiveEnabled;
+  }
 }

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

@@ -382,7 +382,8 @@ public FSDataOutputStream create(final Path f,
     try {
       TracingContext tracingContext = new TracingContext(clientCorrelationId,
           fileSystemId, FSOperationType.CREATE, overwrite, tracingHeaderFormat, listener);
-      OutputStream outputStream = getAbfsStore().createFile(qualifiedPath, statistics, overwrite,
+      OutputStream outputStream = getAbfsStore().createFile(qualifiedPath, statistics,
+          overwrite,
           permission == null ? FsPermission.getFileDefault() : permission,
           FsPermission.getUMask(getConf()), tracingContext);
       statIncrement(FILES_CREATED);
@@ -406,18 +407,23 @@ public FSDataOutputStream createNonRecursive(final Path f, final FsPermission pe
           ERR_CREATE_ON_ROOT,
           null);
     }
-    final Path parent = f.getParent();
     TracingContext tracingContext = new TracingContext(clientCorrelationId,
         fileSystemId, FSOperationType.CREATE_NON_RECURSIVE, tracingHeaderFormat,
         listener);
-    final FileStatus parentFileStatus = tryGetFileStatus(parent, tracingContext);
 
-    if (parentFileStatus == null) {
-      throw new FileNotFoundException("Cannot create file "
-          + f.getName() + " because parent folder does not exist.");
+    try {
+      Path qualifiedPath = makeQualified(f);
+      OutputStream outputStream = getAbfsStore().createPathNonRecursive(
+          qualifiedPath, statistics,
+          overwrite,
+          permission == null ? FsPermission.getFileDefault() : permission,
+          FsPermission.getUMask(getConf()), tracingContext);
+      statIncrement(FILES_CREATED);
+      return new FSDataOutputStream(outputStream, statistics);
+    } catch (AzureBlobFileSystemException ex) {
+      checkException(f, ex);
+      return null;
     }
-
-    return create(f, permission, overwrite, bufferSize, replication, blockSize, progress);
   }
 
   @Override

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

@@ -600,6 +600,24 @@ public void deleteFilesystem(TracingContext tracingContext)
     }
   }
 
+  public OutputStream createPathNonRecursive(final Path path,
+      final FileSystem.Statistics statistics,
+      final boolean overwrite,
+      final FsPermission permission,
+      final FsPermission umask,
+      TracingContext tracingContext)
+      throws IOException {
+    boolean isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
+    AbfsRestOperation op = getClient().createNonRecursivePath(
+        path.toUri().getPath(), true, overwrite,
+        new Permissions(isNamespaceEnabled, permission,
+            umask),
+        false,
+        null, null, tracingContext, isNamespaceEnabled);
+    return createAbfsOutputStreamInstance(statistics, tracingContext,
+        path.toUri().getPath(), false, null, op);
+  }
+
   public OutputStream createFile(final Path path,
       final FileSystem.Statistics statistics, final boolean overwrite,
       final FsPermission permission, final FsPermission umask,
@@ -658,31 +676,45 @@ public OutputStream createFile(final Path path,
 
       }
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
-
-      AbfsLease lease = maybeCreateLease(relativePath, tracingContext);
-      String eTag = extractEtagHeader(op.getResult());
-      return new AbfsOutputStream(
-          populateAbfsOutputStreamContext(
-              isAppendBlob,
-              lease,
-              getClientHandler(),
-              statistics,
-              relativePath,
-              0,
-              eTag,
-              contextEncryptionAdapter,
-              tracingContext));
+      return createAbfsOutputStreamInstance(statistics, tracingContext,
+          relativePath,
+          isAppendBlob, contextEncryptionAdapter, op);
     }
   }
 
+  private AbfsOutputStream createAbfsOutputStreamInstance(final FileSystem.Statistics statistics,
+      final TracingContext tracingContext,
+      final String relativePath,
+      final boolean isAppendBlob,
+      final ContextEncryptionAdapter contextEncryptionAdapter,
+      final AbfsRestOperation op) throws IOException {
+
+    AbfsLease lease = maybeCreateLease(relativePath, tracingContext);
+    String eTag = extractEtagHeader(op.getResult());
+    return new AbfsOutputStream(
+        populateAbfsOutputStreamContext(
+            isAppendBlob,
+            lease,
+            getClientHandler(),
+            statistics,
+            relativePath,
+            0,
+            eTag,
+            contextEncryptionAdapter,
+            tracingContext));
+  }
+
   /**
    * Conditional create overwrite flow ensures that create overwrites is done
    * only if there is match for eTag of existing file.
+   *
    * @param relativePath
    * @param statistics
    * @param permissions contains permission and umask
    * @param isAppendBlob
+   *
    * @return
+   *
    * @throws AzureBlobFileSystemException
    */
   private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePath,
@@ -698,7 +730,8 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
       // avoided for cases when no pre-existing file is present (major portion
       // of create file traffic falls into the case of no pre-existing file).
       op = createClient.createPath(relativePath, true, false, permissions,
-          isAppendBlob, null, contextEncryptionAdapter, tracingContext, getIsNamespaceEnabled(tracingContext));
+          isAppendBlob, null, contextEncryptionAdapter,
+          tracingContext, getIsNamespaceEnabled(tracingContext));
 
     } catch (AbfsRestOperationException e) {
       if (e.getStatusCode() == HttpURLConnection.HTTP_CONFLICT) {
@@ -722,7 +755,8 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
         try {
           // overwrite only if eTag matches with the file properties fetched befpre
           op = createClient.createPath(relativePath, true, true, permissions,
-              isAppendBlob, eTag, contextEncryptionAdapter, tracingContext, getIsNamespaceEnabled(tracingContext));
+              isAppendBlob, eTag, contextEncryptionAdapter,
+              tracingContext, getIsNamespaceEnabled(tracingContext));
         } catch (AbfsRestOperationException ex) {
           if (ex.getStatusCode() == HttpURLConnection.HTTP_PRECON_FAILED) {
             // Is a parallel access case, as file with eTag was just queried
@@ -827,7 +861,8 @@ public void createDirectory(final Path path, final FsPermission permission,
       Permissions permissions = new Permissions(isNamespaceEnabled,
           permission, umask);
       final AbfsRestOperation op = createClient.createPath(getRelativePath(path),
-          false, overwrite, permissions, false, null, null, tracingContext, isNamespaceEnabled);
+          false, overwrite, permissions, false, null, null,
+          tracingContext, isNamespaceEnabled);
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
     }
   }

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

@@ -357,9 +357,17 @@ 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}*/
   public static final String FS_AZURE_BLOB_DIR_DELETE_MAX_THREAD = "fs.azure.blob.dir.delete.max.thread";
+  /**Define if lease to be taken on parent of atomic-directory on non-recursive create path {@value }.*/
+  public static final String FS_AZURE_LEASE_CREATE_NON_RECURSIVE = "fs.azure.lease.create.non.recursive";
   private ConfigurationKeys() {}
 }

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

@@ -36,7 +36,7 @@ public final class FileSystemConfigurations {
   public static final boolean DEFAULT_FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED = true;
   public static final String USER_HOME_DIRECTORY_PREFIX = "/user";
 
-  private static final int SIXTY_SECONDS = 60_000;
+  public static final int SIXTY_SECONDS = 60_000;
 
   // Retry parameter defaults.
   public static final int DEFAULT_MIN_BACKOFF_INTERVAL = 3_000;  // 3s
@@ -170,10 +170,12 @@ 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 boolean DEFAULT_FS_AZURE_LEASE_CREATE_NON_RECURSIVE = false;
   public static final int BLOCK_ID_LENGTH = 60;
   private FileSystemConfigurations() {}
 }