AbfsoutputStream refactor

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

@@ -732,6 +732,10 @@ public boolean isSmallWriteOptimizationEnabled() {
     return this.enableSmallWriteOptimization;
   }
 
+  public void setSmallWriteOptimization(final boolean enableSmallWriteOptimization) {
+    this.enableSmallWriteOptimization = enableSmallWriteOptimization;
+  }
+
   public boolean readSmallFilesCompletely() {
     return this.readSmallFilesCompletely;
   }

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

@@ -45,6 +45,7 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.enums.Trilean;
 import org.apache.hadoop.fs.impl.BackReference;
 import org.apache.hadoop.security.ProviderUtils;
 import org.apache.hadoop.util.Preconditions;
@@ -222,9 +223,7 @@ public void initialize(URI uri, Configuration configuration)
         }
       }
     }
-
     abfsStore.validateConfiguredServiceType(getInitTracingContext());
-
     /*
      * Non-hierarchical-namespace account can not have a customer-provided-key(CPK).
      * Fail initialization of filesystem if the configs are provided. CPK is of

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

@@ -28,12 +28,6 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.URL;
-import java.nio.ByteBuffer;
-import java.nio.CharBuffer;
-import java.nio.charset.CharacterCodingException;
-import java.nio.charset.Charset;
-import java.nio.charset.CharsetDecoder;
-import java.nio.charset.CharsetEncoder;
 import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.time.Instant;
@@ -146,6 +140,7 @@
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
 import org.apache.http.client.utils.URIBuilder;
 
+import static java.net.HttpURLConnection.HTTP_CONFLICT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.METADATA_INCOMPLETE_RENAME_FAILURES;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.RENAME_RECOVERY;
@@ -168,6 +163,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_BLOB_DOMAIN_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_DFS_DOMAIN_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT;
+import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.PATH_EXISTS;
 
 /**
  * Provides the bridging logic between Hadoop's abstract filesystem and Azure Storage.
@@ -297,16 +293,16 @@ public AzureBlobFileSystemStore(
   public void validateConfiguredServiceType(TracingContext tracingContext)
       throws AzureBlobFileSystemException {
     // Todo: [FnsOverBlob] - Remove this check, Failing FS Init with Blob Endpoint Until FNS over Blob is ready.
-    if (getConfiguredServiceType() == AbfsServiceType.BLOB) {
-      throw new InvalidConfigurationValueException(FS_DEFAULT_NAME_KEY);
-    }
-    if (getIsNamespaceEnabled(tracingContext) && getConfiguredServiceType() == AbfsServiceType.BLOB) {
-      // This could be because of either wrongly configured url or wrongly configured fns service type.
-      if (identifyAbfsServiceTypeFromUrl() == AbfsServiceType.BLOB) {
-        throw new InvalidConfigurationValueException(FS_DEFAULT_NAME_KEY, "Wrong Domain Suffix for HNS Account");
-      }
-      throw new InvalidConfigurationValueException(FS_AZURE_FNS_ACCOUNT_SERVICE_TYPE, "Wrong Service Type for HNS Accounts");
-    }
+//    if (getConfiguredServiceType() == AbfsServiceType.BLOB) {
+//      throw new InvalidConfigurationValueException(FS_DEFAULT_NAME_KEY);
+//    }
+//    if (getIsNamespaceEnabled(tracingContext) && getConfiguredServiceType() == AbfsServiceType.BLOB) {
+//      // This could be because of either wrongly configured url or wrongly configured fns service type.
+//      if (identifyAbfsServiceTypeFromUrl() == AbfsServiceType.BLOB) {
+//        throw new InvalidConfigurationValueException(FS_DEFAULT_NAME_KEY, "Wrong Domain Suffix for HNS Account");
+//      }
+//      throw new InvalidConfigurationValueException(FS_AZURE_FNS_ACCOUNT_SERVICE_TYPE, "Wrong Service Type for HNS Accounts");
+//    }
   }
 
   /**
@@ -636,9 +632,10 @@ public OutputStream createFile(final Path path,
       final FsPermission permission, final FsPermission umask,
       TracingContext tracingContext) throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("createFile", "createPath")) {
+      AbfsClient createClient = getClient(abfsConfiguration.getIngressServiceType());
       boolean isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
       LOG.debug("createFile filesystem: {} path: {} overwrite: {} permission: {} umask: {} isNamespaceEnabled: {}",
-              client.getFileSystem(),
+              createClient.getFileSystem(),
               path,
               overwrite,
               permission,
@@ -650,6 +647,11 @@ public OutputStream createFile(final Path path,
       if (isAppendBlobKey(path.toString())) {
         isAppendBlob = true;
       }
+      if (!isNamespaceEnabled) {
+        if (createClient instanceof AbfsBlobClient && path.getParent() != null && !path.getParent().isRoot()) {
+          createDirectory(path.getParent(), permission, umask, tracingContext);
+        }
+      }
 
       // if "fs.azure.enable.conditional.create.overwrite" is enabled and
       // is a create request with overwrite=true, create will follow different
@@ -661,9 +663,9 @@ public OutputStream createFile(final Path path,
       }
 
       final ContextEncryptionAdapter contextEncryptionAdapter;
-      if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
+      if (createClient.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
         contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
-            client.getEncryptionContextProvider(), getRelativePath(path));
+            createClient.getEncryptionContextProvider(), getRelativePath(path));
       } else {
         contextEncryptionAdapter = NoContextEncryptionAdapter.getInstance();
       }
@@ -674,11 +676,12 @@ public OutputStream createFile(final Path path,
             new Permissions(isNamespaceEnabled, permission, umask),
             isAppendBlob,
             contextEncryptionAdapter,
-            tracingContext
+            tracingContext,
+            createClient
         );
 
       } else {
-        op = client.createPath(relativePath, true,
+        op = createClient.createPath(relativePath, true,
             overwrite,
             new Permissions(isNamespaceEnabled, permission, umask),
             isAppendBlob,
@@ -690,20 +693,33 @@ public OutputStream createFile(final Path path,
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
 
       AbfsLease lease = maybeCreateLease(relativePath, tracingContext);
-
+      String eTag = op.getResult().getResponseHeader(HttpHeaderConfigurations.ETAG);
+      checkSmallWriteOptimization();
       return new AbfsOutputStream(
           populateAbfsOutputStreamContext(
               isAppendBlob,
               lease,
-              client,
+              clientHandler,
               statistics,
               relativePath,
               0,
+              eTag,
               contextEncryptionAdapter,
               tracingContext));
     }
   }
 
+  /**
+   * Checks and disables the small write optimization if the client is an instance of {@link AbfsBlobClient}
+   * and the small write optimization is enabled in the configuration.
+   */
+  private void checkSmallWriteOptimization() {
+    AbfsClient ingressClient = getClient(abfsConfiguration.getIngressServiceType());
+    if (ingressClient instanceof AbfsBlobClient && abfsConfiguration.isSmallWriteOptimizationEnabled()) {
+      abfsConfiguration.setSmallWriteOptimization(false);
+    }
+  }
+
   /**
    * Conditional create overwrite flow ensures that create overwrites is done
    * only if there is match for eTag of existing file.
@@ -719,14 +735,14 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
       final Permissions permissions,
       final boolean isAppendBlob,
       final ContextEncryptionAdapter contextEncryptionAdapter,
-      final TracingContext tracingContext) throws IOException {
+      final TracingContext tracingContext, AbfsClient createClient) throws IOException {
     AbfsRestOperation op;
 
     try {
       // Trigger a create with overwrite=false first so that eTag fetch can be
       // 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 = client.createPath(relativePath, true, false, permissions,
+      op = createClient.createPath(relativePath, true, false, permissions,
           isAppendBlob, null, contextEncryptionAdapter, tracingContext);
 
     } catch (AbfsRestOperationException e) {
@@ -751,7 +767,7 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
 
         try {
           // overwrite only if eTag matches with the file properties fetched befpre
-          op = client.createPath(relativePath, true, true, permissions,
+          op = createClient.createPath(relativePath, true, true, permissions,
               isAppendBlob, eTag, contextEncryptionAdapter, tracingContext);
         } catch (AbfsRestOperationException ex) {
           if (ex.getStatusCode() == HttpURLConnection.HTTP_PRECON_FAILED) {
@@ -779,7 +795,7 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
    *
    * @param isAppendBlob   is Append blob support enabled?
    * @param lease          instance of AbfsLease for this AbfsOutputStream.
-   * @param client         AbfsClient.
+   * @param clientHandler  AbfsClientHandler.
    * @param statistics     FileSystem statistics.
    * @param path           Path for AbfsOutputStream.
    * @param position       Position or offset of the file being opened, set to 0
@@ -791,10 +807,11 @@ private AbfsRestOperation conditionalCreateOverwriteFile(final String relativePa
   private AbfsOutputStreamContext populateAbfsOutputStreamContext(
       boolean isAppendBlob,
       AbfsLease lease,
-      AbfsClient client,
+      AbfsClientHandler clientHandler,
       FileSystem.Statistics statistics,
       String path,
       long position,
+      String eTag,
       ContextEncryptionAdapter contextEncryptionAdapter,
       TracingContext tracingContext) {
     int bufferSize = abfsConfiguration.getWriteBufferSize();
@@ -815,24 +832,28 @@ private AbfsOutputStreamContext populateAbfsOutputStreamContext(
             .withEncryptionAdapter(contextEncryptionAdapter)
             .withBlockFactory(getBlockFactory())
             .withBlockOutputActiveBlocks(blockOutputActiveBlocks)
-            .withClient(client)
+            .withClientHandler(clientHandler)
             .withPosition(position)
             .withFsStatistics(statistics)
             .withPath(path)
             .withExecutorService(new SemaphoredDelegatingExecutor(boundedThreadPool,
                 blockOutputActiveBlocks, true))
             .withTracingContext(tracingContext)
             .withAbfsBackRef(fsBackRef)
+            .withIngressServiceType(abfsConfiguration.getIngressServiceType())
+            .withDFSToBlobFallbackEnabled(abfsConfiguration.isDfsToBlobFallbackEnabled())
+            .withETag(eTag)
             .build();
   }
 
   public void createDirectory(final Path path, final FsPermission permission,
       final FsPermission umask, TracingContext tracingContext)
       throws IOException {
     try (AbfsPerfInfo perfInfo = startTracking("createDirectory", "createPath")) {
+      AbfsClient createClient = clientHandler.getClient(abfsConfiguration.getIngressServiceType());
       boolean isNamespaceEnabled = getIsNamespaceEnabled(tracingContext);
       LOG.debug("createDirectory filesystem: {} path: {} permission: {} umask: {} isNamespaceEnabled: {}",
-              client.getFileSystem(),
+              createClient.getFileSystem(),
               path,
               permission,
               umask,
@@ -842,7 +863,12 @@ public void createDirectory(final Path path, final FsPermission permission,
           !isNamespaceEnabled || abfsConfiguration.isEnabledMkdirOverwrite();
       Permissions permissions = new Permissions(isNamespaceEnabled,
           permission, umask);
-      final AbfsRestOperation op = client.createPath(getRelativePath(path),
+      if (!isNamespaceEnabled && createClient instanceof AbfsBlobClient) {
+        ((AbfsBlobClient) createClient).createMarkerBlobs(path, overwrite, permissions, false,
+            null,
+            null, tracingContext);
+      }
+      final AbfsRestOperation op = createClient.createPath(getRelativePath(path),
           false, overwrite, permissions, false, null, null, tracingContext);
       perfInfo.registerResult(op.getResult()).registerSuccess(true);
     }
@@ -978,6 +1004,7 @@ public OutputStream openFileForWrite(final Path path,
               overwrite);
 
       String relativePath = getRelativePath(path);
+      AbfsClient writeClient = getClient(abfsConfiguration.getIngressServiceType());
 
       final AbfsRestOperation op = client
           .getPathStatus(relativePath, false, tracingContext, null);
@@ -990,7 +1017,7 @@ public OutputStream openFileForWrite(final Path path,
         throw new AbfsRestOperationException(
                 AzureServiceErrorCode.PATH_NOT_FOUND.getStatusCode(),
                 AzureServiceErrorCode.PATH_NOT_FOUND.getErrorCode(),
-                "openFileForRead must be used with files and not directories",
+                "openFileForWrite must be used with files and not directories",
                 null);
       }
 
@@ -1004,8 +1031,9 @@ public OutputStream openFileForWrite(final Path path,
       }
 
       AbfsLease lease = maybeCreateLease(relativePath, tracingContext);
+      final String eTag = extractEtagHeader(op.getResult());
       final ContextEncryptionAdapter contextEncryptionAdapter;
-      if (client.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
+      if (writeClient.getEncryptionType() == EncryptionType.ENCRYPTION_CONTEXT) {
         final String encryptionContext = op.getResult()
             .getResponseHeader(
                 HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT);
@@ -1014,20 +1042,21 @@ public OutputStream openFileForWrite(final Path path,
               "File doesn't have encryptionContext.");
         }
         contextEncryptionAdapter = new ContextProviderEncryptionAdapter(
-            client.getEncryptionContextProvider(), getRelativePath(path),
+            writeClient.getEncryptionContextProvider(), getRelativePath(path),
             encryptionContext.getBytes(StandardCharsets.UTF_8));
       } else {
         contextEncryptionAdapter = NoContextEncryptionAdapter.getInstance();
       }
-
+      checkSmallWriteOptimization();
       return new AbfsOutputStream(
           populateAbfsOutputStreamContext(
               isAppendBlob,
               lease,
-              client,
+              clientHandler,
               statistics,
               relativePath,
               offset,
+              eTag,
               contextEncryptionAdapter,
               tracingContext));
     }
@@ -2156,6 +2185,11 @@ public AbfsClient getClient(AbfsServiceType serviceType) {
     return this.clientHandler.getClient(serviceType);
   }
 
+  @VisibleForTesting
+  public AbfsClientHandler getClientHandler() {
+    return clientHandler;
+  }
+
   @VisibleForTesting
   void setClient(AbfsClient client) {
     this.client = client;

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

@@ -60,6 +60,7 @@ public final class AbfsHttpConstants {
   public static final String BLOCK_BLOB_TYPE = "BlockBlob";
   public static final String BLOCK_TYPE_COMMITTED = "committed";
   public static final String TOKEN_VERSION = "2";
+  public static final String APPEND_BLOCK = "appendblock";
 
   public static final String JAVA_VENDOR = "java.vendor";
   public static final String JAVA_VERSION = "java.version";
@@ -149,7 +150,7 @@ public final class AbfsHttpConstants {
   public enum ApiVersion {
 
     DEC_12_2019("2019-12-12"),
-    APR_10_2021("2021-04-10"),
+    AUG_06_2021("2021-08-06"),
     AUG_03_2023("2023-08-03");
 
     private final String xMsApiVersion;
@@ -213,7 +214,11 @@ public static ApiVersion getCurrentVersion() {
   public static final String XML_TAG_BLOB_ERROR_MESSAGE_END_XML = "</Message>";
   public static final String XML_TAG_COMMITTED_BLOCKS = "CommittedBlocks";
   public static final String XML_TAG_BLOCK_NAME = "Block";
-
+  public static final String XML_VERSION = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n";
+  public static final String BLOCK_LIST_START_TAG = "<BlockList>\n";
+  public static final String BLOCK_LIST_END_TAG = "</BlockList>\n";
+  public static final String LATEST_BLOCK_FORMAT = "<Latest>%s</Latest>\n";
+  public static final String PUT_BLOCK_LIST = "PutBlockList";
   /**
    * List of configurations that are related to Customer-Provided-Keys.
    * <ol>

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

@@ -168,5 +168,7 @@ public final class FileSystemConfigurations {
   public static final int HUNDRED = 100;
   public static final long THOUSAND = 1000L;
 
+  public static final int BLOCK_ID_LENGTH = 60;
+
   private FileSystemConfigurations() {}
 }