HADOOP-19233: ABFS: [FnsOverBlob] Implementing Rename and Delete APIs over Blob Endpoint

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

@@ -447,15 +447,21 @@ public FSDataOutputStream createNonRecursive(final Path f,
           ERR_CREATE_ON_ROOT,
           null);
     }
-
-    TracingContext tracingContext = new TracingContext(clientCorrelationId,
-        fileSystemId, FSOperationType.CREATE_NON_RECURSIVE, tracingHeaderFormat,
-        listener);
-
     Path qualifiedPath = makeQualified(f);
-    getAbfsStore().createNonRecursivePreCheck(qualifiedPath, tracingContext);
-    return create(f, permission, overwrite, bufferSize, replication,
-        blockSize, progress);
+    try {
+      TracingContext tracingContext = new TracingContext(clientCorrelationId,
+          fileSystemId, FSOperationType.CREATE_NON_RECURSIVE, tracingHeaderFormat,
+          listener);
+      OutputStream outputStream =  getAbfsStore().createNonRecursive(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;
+    }
   }
 
   @Override
@@ -535,34 +541,20 @@ public boolean rename(final Path src, final Path dst) throws IOException {
       return tryGetFileStatus(qualifiedSrcPath, tracingContext) != null;
     }
 
-    FileStatus dstFileStatus = null;
+    FileStatus dstFileStatus = tryGetFileStatus(qualifiedDstPath, tracingContext);
+    Path adjustedDst = dst;
     if (qualifiedSrcPath.equals(qualifiedDstPath)) {
-      // rename to itself
-      // - if it doesn't exist, return false
-      // - if it is file, return true
-      // - if it is dir, return false.
-      dstFileStatus = tryGetFileStatus(qualifiedDstPath, tracingContext);
-      if (dstFileStatus == null) {
+      return dstFileStatus != null && !dstFileStatus.isDirectory();
+    } else if (dstFileStatus != null) {
+      if (!dstFileStatus.isDirectory()) {
         return false;
       }
-      return dstFileStatus.isDirectory() ? false : true;
-    }
-
-    // Non-HNS account need to check dst status on driver side.
-    if (!getIsNamespaceEnabled(tracingContext) && dstFileStatus == null) {
-      dstFileStatus = tryGetFileStatus(qualifiedDstPath, tracingContext);
+      String sourceFileName = src.getName();
+      adjustedDst = new Path(dst, sourceFileName);
     }
 
     try {
-      String sourceFileName = src.getName();
-      Path adjustedDst = dst;
-
-      if (dstFileStatus != null) {
-        adjustedDst = new Path(dst, sourceFileName);
-      }
-
       qualifiedDstPath = makeQualified(adjustedDst);
-
       getAbfsStore().rename(qualifiedSrcPath, qualifiedDstPath, tracingContext,
           null);
       return true;

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

@@ -139,7 +139,6 @@
 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.azurebfs.AbfsStatistic.METADATA_INCOMPLETE_RENAME_FAILURES;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.RENAME_RECOVERY;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CHAR_EQUALS;
@@ -161,7 +160,6 @@
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.INFINITE_LEASE_DURATION;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemUriSchemes.ABFS_BLOB_DOMAIN_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations.X_MS_ENCRYPTION_CONTEXT;
-import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_CREATE_ON_ROOT;
 import static org.apache.hadoop.fs.azurebfs.utils.UriUtils.isKeyForDirectorySet;
 
 /**
@@ -649,18 +647,16 @@
    *
    * @throws IOException if there is an issue with the operation.
    */
-  public void createNonRecursivePreCheck(final Path path,
+  public OutputStream createNonRecursive(final Path path,
+      final FileSystem.Statistics statistics, final boolean overwrite,
+      final FsPermission permission, final FsPermission umask,
       TracingContext tracingContext)
       throws IOException {
-    if (path.getParent() != null) {
-      getClient().createNonRecursivePreCheck(path.getParent(),
-          tracingContext);
-    } else {
-      throw new AbfsRestOperationException(HTTP_CONFLICT,
-          AzureServiceErrorCode.PATH_CONFLICT.getErrorCode(),
-          ERR_CREATE_ON_ROOT,
-          null);
-    }
+    getClient().createNonRecursivePreCheck(path.getParent(),
+        tracingContext);
+   return createFile(path, statistics, overwrite, permission,
+       umask, tracingContext);
+
   }
 
   public OutputStream createFile(final Path path,

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

@@ -80,7 +80,6 @@
 import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
 import static java.net.HttpURLConnection.HTTP_OK;
 import static java.net.HttpURLConnection.HTTP_PRECON_FAILED;
-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;
@@ -447,8 +446,6 @@ public void createNonRecursivePreCheck(Path parentPath,
             + " because parent folder does not exist.");
       }
       throw ex;
-    } finally {
-      getAbfsCounters().incrementCounter(CALL_GET_FILE_STATUS, 1);
     }
   }
 

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

@@ -203,7 +203,7 @@ private AbfsClient(final URL baseUrl,
 
     if (encryptionContextProvider != null) {
       this.encryptionContextProvider = encryptionContextProvider;
-      xMsVersion = ApiVersion.APR_10_2021; // will be default once server change deployed
+      xMsVersion = ApiVersion.AUG_03_2023; // will be default once server change deployed
       encryptionType = EncryptionType.ENCRYPTION_CONTEXT;
     } else if (abfsConfiguration.getEncodedClientProvidedEncryptionKey() != null) {
       clientProvidedEncryptionKey =

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

@@ -170,7 +170,7 @@ private List<Path> dequeue() {
     return pathListForConsumption;
   }
 
-  private synchronized int size() {
+  synchronized int size() {
     return pathQueue.size();
   }
 

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

@@ -63,10 +63,7 @@
 import org.apache.hadoop.test.LambdaTestUtils;
 import org.apache.hadoop.util.Lists;
 
-import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.BLOCK_LIST_END_TAG;
-import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.BLOCK_LIST_START_TAG;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.CPK_IN_NON_HNS_ACCOUNT_ERROR_MESSAGE;
-import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.XML_VERSION;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENCRYPTION_CONTEXT_PROVIDER_TYPE;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENCRYPTION_ENCODED_CLIENT_PROVIDED_KEY;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENCRYPTION_ENCODED_CLIENT_PROVIDED_KEY_SHA;
@@ -77,6 +74,7 @@
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.ENCRYPTION_KEY_LEN;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT;
 import static org.apache.hadoop.fs.azurebfs.contracts.services.AppendRequestParameters.Mode.APPEND_MODE;
+import static org.apache.hadoop.fs.azurebfs.services.AbfsBlobClient.generateBlockListXml;
 import static org.apache.hadoop.fs.azurebfs.utils.AclTestHelpers.aclEntry;
 import static org.apache.hadoop.fs.azurebfs.utils.EncryptionType.ENCRYPTION_CONTEXT;
 import static org.apache.hadoop.fs.azurebfs.utils.EncryptionType.GLOBAL_KEY;
@@ -191,12 +189,6 @@ public void testCustomEncryptionCombinations() throws Exception {
     }
   }
 
-  protected static String generateBlockListXml() {
-    return XML_VERSION
-        + BLOCK_LIST_START_TAG
-        + BLOCK_LIST_END_TAG;
-  }
-
   private void validateCpkResponseHeadersForCombination(final AzureBlobFileSystem fs)
       throws Exception {
     Path testPath = path("/testFile");
@@ -328,7 +320,7 @@ private AbfsRestOperation callOperation(AzureBlobFileSystem fs,
           return ingressClient.flush(path, 3, false, false, null,
               null, encryptionAdapter, getTestTracingContext(fs, false));
         } else {
-          byte[] buffer = generateBlockListXml().getBytes(StandardCharsets.UTF_8);
+          byte[] buffer = generateBlockListXml(new ArrayList<>()).getBytes(StandardCharsets.UTF_8);
           return ingressClient.flush(buffer, path, false, null,
               null, null, encryptionAdapter, getTestTracingContext(fs, false));
         }

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

@@ -345,16 +345,6 @@ public void deleteBlobDirParallelThreadToDeleteOnDifferentTracingContext()
     fs.close();
   }
 
-  /**
-   * Assumes that the AzureBlobFileSystem's client is an instance of `AbfsBlobClient`.
-   * This assumption is used to ensure that the test runs only if the correct client type is present.
-   *
-   * @throws IOException if the file system client cannot be retrieved
-   */
-  private void assumeBlobClient() throws IOException {
-    Assume.assumeTrue(getFileSystem().getAbfsClient() instanceof AbfsBlobClient);
-  }
-
   /**
    * Tests deleting an implicit directory and its contents. The test verifies that after deletion,
    * both the directory and its child file no longer exist.
@@ -364,7 +354,7 @@ private void assumeBlobClient() throws IOException {
   @Test
   public void testDeleteImplicitDir() throws Exception {
     AzureBlobFileSystem fs = getFileSystem();
-    assumeBlobClient();
+    assumeBlobServiceType();
     fs.mkdirs(new Path("/testDir/dir1"));
     fs.create(new Path("/testDir/dir1/file1"));
     AbfsBlobClient client = (AbfsBlobClient) fs.getAbfsClient();
@@ -389,7 +379,7 @@ public void testDeleteImplicitDir() throws Exception {
   public void testDeleteImplicitDirWithSingleListResults() throws Exception {
     AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(
             getRawConfiguration());
-    assumeBlobClient();
+    assumeBlobServiceType();
     AbfsBlobClient client = (AbfsBlobClient) fs.getAbfsClient();
     AbfsBlobClient spiedClient = Mockito.spy(client);
     fs.getAbfsStore().setClient(spiedClient);
@@ -425,7 +415,7 @@ public void testDeleteImplicitDirWithSingleListResults() throws Exception {
   @Test
   public void testDeleteExplicitDirInImplicitParentDir() throws Exception {
     AzureBlobFileSystem fs = getFileSystem();
-    assumeBlobClient();
+    assumeBlobServiceType();
     AbfsBlobClient client = (AbfsBlobClient) fs.getAbfsClient();
     fs.mkdirs(new Path("/testDir/dir1"));
     fs.create(new Path("/testDir/dir1/file1"));
@@ -453,7 +443,7 @@ public void testDeleteExplicitDirInImplicitParentDir() throws Exception {
   @Test
   public void testDeleteParallelBlobFailure() throws Exception {
     AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
-    assumeBlobClient();
+    assumeBlobServiceType();
     AbfsBlobClient client = Mockito.spy((AbfsBlobClient) fs.getAbfsClient());
     AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());
     store.setClient(client);
@@ -497,7 +487,7 @@ public void testDeleteRootWithNonRecursion() throws Exception {
    */
   @Test
   public void testProducerStopOnDeleteFailure() throws Exception {
-    assumeBlobClient();
+    assumeBlobServiceType();
     Configuration configuration = Mockito.spy(getRawConfiguration());
     AzureBlobFileSystem fs = Mockito.spy(
             (AzureBlobFileSystem) FileSystem.get(configuration));
@@ -575,7 +565,7 @@ public void testProducerStopOnDeleteFailure() throws Exception {
   @Test
   public void testDeleteEmitDeletionCountInClientRequestId() throws Exception {
     AzureBlobFileSystem fs = Mockito.spy(getFileSystem());
-    assumeBlobClient();
+    assumeBlobServiceType();
     AbfsBlobClient client = (AbfsBlobClient) fs.getAbfsClient();
     AbfsBlobClient spiedClient = Mockito.spy(client);
     AzureBlobFileSystemStore store = Mockito.spy(fs.getAbfsStore());

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

@@ -58,6 +58,7 @@ public class ITestAzureBlobFileSystemE2E extends AbstractAbfsIntegrationTest {
   private static final int TEST_STABLE_DEFAULT_CONNECTION_TIMEOUT_MS = 500;
   private static final int TEST_STABLE_DEFAULT_READ_TIMEOUT_MS = 30000;
   private static final int TEST_UNSTABLE_READ_TIMEOUT_MS = 1;
+  private static final int TEST_WRITE_BYTE_VALUE = 20;
 
   public ITestAzureBlobFileSystemE2E() throws Exception {
     super();
@@ -234,7 +235,7 @@ public void testFlushWithFileNotFoundException() throws Exception {
 
       fs.delete(testFilePath, true);
       assertPathDoesNotExist(fs, "This path should not exist", testFilePath);
-
+      stream.write(TEST_WRITE_BYTE_VALUE);
       if (client instanceof AbfsDfsClient) {
         intercept(FileNotFoundException.class, stream::close);
       } else {

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

@@ -28,6 +28,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.constants.AbfsServiceType;
 import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
 import org.apache.hadoop.fs.azurebfs.constants.HttpOperationType;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsDriverException;
@@ -260,6 +261,9 @@ public void testWriteAfterBreakLease() throws Exception {
 
     LambdaTestUtils.intercept(IOException.class, client instanceof AbfsBlobClient
         ? ERR_LEASE_EXPIRED_BLOB : ERR_LEASE_EXPIRED, () -> {
+      if (isAppendBlobEnabled() && getIngressServiceType() == AbfsServiceType.BLOB) {
+        out.write(20);
+      }
       out.close();
       return "Expected exception on close after lease break but got " + out;
     });

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

@@ -28,6 +28,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
+import org.junit.Assume;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.mockito.stubbing.Stubber;
@@ -38,6 +39,7 @@
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.constants.AbfsServiceType;
 import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
 import org.apache.hadoop.fs.azurebfs.constants.HttpHeaderConfigurations;
 import org.apache.hadoop.fs.azurebfs.contracts.services.DfsListResultEntrySchema;
@@ -122,6 +124,7 @@ public Void call() throws Exception {
    */
   @Test
   public void testListPathTracingContext() throws Exception {
+    Assume.assumeTrue(getAbfsServiceType() == AbfsServiceType.DFS);
     final AzureBlobFileSystem fs = getFileSystem();
     final AzureBlobFileSystem spiedFs = Mockito.spy(fs);
     final AzureBlobFileSystemStore spiedStore = Mockito.spy(fs.getAbfsStore());

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

@@ -24,8 +24,8 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import org.apache.hadoop.fs.azurebfs.constants.AbfsServiceType;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
-import org.apache.hadoop.fs.azurebfs.services.AbfsBlobClient;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathIOException;
@@ -90,7 +90,7 @@ public void testRenameFileUsingUnicode() throws Exception {
     assertIsFile(fs, filePath);
 
     Path folderPath2 = new Path(destDir);
-    if (getFileSystem().getAbfsClient() instanceof AbfsBlobClient
+    if (getAbfsServiceType() == AbfsServiceType.BLOB
         && destDir.contains(COLON)) {
       AbfsRestOperationException ex = intercept(
           AbfsRestOperationException.class, () -> {

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

@@ -17,12 +17,13 @@
  */
 package org.apache.hadoop.fs.azurebfs;
 
+import java.io.IOException;
+
 import org.junit.Assume;
 import org.junit.Test;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
-import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.azurebfs.services.AuthType;
 
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION;
@@ -40,23 +41,19 @@ public void testWithWrongSharedKey() throws Exception {
     Assume.assumeTrue(this.getAuthType() == AuthType.SharedKey);
     Configuration config = this.getRawConfiguration();
     config.setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION,
-        false);
+        true);
     String accountName = this.getAccountName();
     String configkKey = FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME + "." + accountName;
     // a wrong sharedKey
     String secret = "XjUjsGherkDpljuyThd7RpljhR6uhsFjhlxRpmhgD12lnj7lhfRn8kgPt5"
         + "+MJHS7UJNDER+jn6KP6Jnm2ONQlm==";
     config.set(configkKey, secret);
-
-    AbfsClient abfsClient = this.getFileSystem(config).getAbfsClient();
-    intercept(AbfsRestOperationException.class,
+    intercept(IOException.class,
         "\"Server failed to authenticate the request. Make sure the value of "
             + "Authorization header is formed correctly including the "
             + "signature.\", 403",
         () -> {
-          abfsClient
-              .getAclStatus("/", getTestTracingContext(getFileSystem(), false));
+          FileSystem.newInstance(config);
         });
   }
-
 }

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

@@ -42,7 +42,6 @@ public ITestAbfsClientHandler() throws Exception{
   public void testAbfsClientHandlerInitialization() throws Exception {
     AzureBlobFileSystem fs = getFileSystem();
     AbfsClientHandler clientHandler = fs.getAbfsStore().getClientHandler();
-    Assertions.assertThat(clientHandler.getClient()).isInstanceOf(AbfsDfsClient.class);
     Assertions.assertThat(clientHandler.getClient(AbfsServiceType.DFS)).isInstanceOf(AbfsDfsClient.class);
     Assertions.assertThat(clientHandler.getClient(AbfsServiceType.BLOB)).isInstanceOf(AbfsBlobClient.class);
   }