HADOOP-19497: [ABFS] Enable rename and create recovery from client transaction id over DFS endpoint (#7509) (#7612)

Contributed by Manish Bhatt
Reiewed by Anmol Asrani, Anuj Modi, Manika Joshi

Signed off by: Anuj Modi<anujmodi@apache.org>

Author: bhattmanish98

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

@@ -199,13 +199,10 @@ public String toString() {
     }
 
     public static ApiVersion getCurrentVersion() {
-      return DEC_12_2019;
+      return NOV_04_2024;
     }
   }
 
-  @Deprecated
-  public static final String DECEMBER_2019_API_VERSION = ApiVersion.DEC_12_2019.toString();
-
   /**
    * List of Constants Used by Blob Endpoint Rest APIs.
    */

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

@@ -198,7 +198,7 @@ public final class FileSystemConfigurations {
 
   public static final int DEFAULT_FS_AZURE_BLOB_DELETE_THREAD = DEFAULT_FS_AZURE_LISTING_ACTION_THREADS;
 
-  public static final boolean DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = false;
+  public static final boolean DEFAULT_FS_AZURE_ENABLE_CLIENT_TRANSACTION_ID = true;
 
   private FileSystemConfigurations() {}
 }

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

@@ -218,7 +218,6 @@ private AbfsClient(final URL baseUrl,
       this.encryptionContextProvider = encryptionContextProvider;
       // Version update needed to support x-ms-encryption-context header
       // @link https://learn.microsoft.com/en-us/rest/api/storageservices/put-block?tabs=microsoft-entra-id}
-      xMsVersion = ApiVersion.AUG_03_2023; // will be default once server change deployed
       encryptionType = EncryptionType.ENCRYPTION_CONTEXT;
     } else if (abfsConfiguration.getEncodedClientProvidedEncryptionKey() != null) {
       clientProvidedEncryptionKey =
@@ -228,11 +227,6 @@ private AbfsClient(final URL baseUrl,
       encryptionType = EncryptionType.GLOBAL_KEY;
     }
 
-    // Version update needed to support x-ms-client-transaction-id header
-    if (abfsConfiguration.getIsClientTransactionIdEnabled()) {
-      xMsVersion = ApiVersion.NOV_04_2024;
-    }
-
     String sslProviderName = null;
 
     if (this.baseUrl.toString().startsWith(HTTPS_SCHEME)) {

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

@@ -72,7 +72,7 @@ public final class AbfsErrors {
   public static final String ERR_CREATE_RECOVERY =
       "Error while recovering from create failure.";
   public static final String ERR_RENAME_RECOVERY =
-      "Error while recovering from rename failure.";
+      "Error while recovering from rename failure for path: ";
   public static final String ERR_BLOB_LIST_PARSING = "Parsing of XML List Response Failed in BlobClient.";
   public static final String ERR_DFS_LIST_PARSING = "Parsing of Json List Response Failed in DfsClient.";
   public static final String INCORRECT_INGRESS_TYPE = "Ingress Type Cannot be DFS for Blob endpoint configured filesystem.";

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

@@ -2289,6 +2289,6 @@ public void answer(final AbfsRestOperation mockedObj,
                   null, op);
             }
           }
-        });
+        }, 0);
   }
 }

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

@@ -2673,12 +2673,13 @@ public void testFailureInGetPathStatusDuringRenameRecovery() throws Exception {
       final String[] clientTransactionId = new String[1];
       mockAddClientTransactionIdToHeader(abfsDfsClient, clientTransactionId);
       mockRetriedRequest(abfsDfsClient, new ArrayList<>());
-      boolean[] flag = new boolean[1];
+      int[] flag = new int[1];
       Mockito.doAnswer(getPathStatus -> {
-        if (!flag[0]) {
-          flag[0] = true;
+        if (flag[0] == 1) {
+          flag[0] += 1;
           throw new AbfsRestOperationException(HTTP_CLIENT_TIMEOUT, "", "", new Exception());
         }
+        flag[0] += 1;
         return getPathStatus.callRealMethod();
       }).when(abfsDfsClient).getPathStatus(
           Mockito.nullable(String.class), Mockito.nullable(Boolean.class),
@@ -2737,6 +2738,6 @@ public void answer(final AbfsRestOperation mockedObj,
                    SOURCE_PATH_NOT_FOUND.getErrorCode(), EMPTY_STRING, null, op);
              }
            }
-         });
+         }, 1);
    }
 }

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

@@ -152,11 +152,11 @@ private boolean isThreadRunning(String threadName) {
    * @throws Exception if an error occurs while mocking the operation creation
    */
   public static void mockAbfsOperationCreation(final AbfsClient abfsClient,
-      final MockIntercept mockIntercept) throws Exception {
-    boolean[] flag = new boolean[1];
+      final MockIntercept mockIntercept, int failedCall) throws Exception {
+    int[] flag = new int[1];
     Mockito.doAnswer(answer -> {
-          if (!flag[0]) {
-            flag[0] = true;
+          if (flag[0] == failedCall) {
+            flag[0] += 1;
             AbfsRestOperation op = Mockito.spy(
                 new AbfsRestOperation(
                     answer.getArgument(0),
@@ -174,6 +174,7 @@ public static void mockAbfsOperationCreation(final AbfsClient abfsClient,
             Mockito.doReturn(true).when(op).isARetriedRequest();
             return op;
           }
+          flag[0] += 1;
           return answer.callRealMethod();
         }).when(abfsClient)
         .getAbfsRestOperation(any(), any(), any(), any());

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

@@ -44,6 +44,7 @@
 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.services.AzureServiceErrorCode;
+import org.apache.hadoop.fs.azurebfs.security.ContextEncryptionAdapter;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.fs.statistics.IOStatistics;
 
@@ -58,7 +59,9 @@
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
 import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.lookupCounterStatistic;
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.mockito.ArgumentMatchers.anyBoolean;
 import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -180,6 +183,12 @@ AbfsClient getMockAbfsClient() throws IOException {
         .when(spyClient)
         .createRenameRestOperation(Mockito.any(URL.class), anyList());
 
+    Mockito.doCallRealMethod()
+        .when(spyClient)
+        .getPathStatus(anyString(), anyBoolean(),
+            Mockito.any(TracingContext.class),
+            Mockito.any(ContextEncryptionAdapter.class));
+
     return spyClient;
 
   }
@@ -275,9 +284,14 @@ public void testRenameRecoveryEtagMatchFsLevel() throws IOException {
     // 4 calls should have happened in total for rename
     // 1 -> original rename rest call, 2 -> first retry,
     // +2 for getPathStatus calls
+    int totalConnections = 4;
+    if (!getConfiguration().getIsClientTransactionIdEnabled()) {
+      // 1 additional getPathStatus call to get dest etag
+      totalConnections++;
+    }
     assertThatStatisticCounter(ioStats,
             CONNECTIONS_MADE.getStatName())
-            .isEqualTo(5 + connMadeBeforeRename);
+            .isEqualTo(totalConnections + connMadeBeforeRename);
     // the RENAME_PATH_ATTEMPTS stat should be incremented by 1
     // retries happen internally within AbfsRestOperation execute()
     // the stat for RENAME_PATH_ATTEMPTS is updated only once before execute() is called
@@ -350,21 +364,18 @@ public void testRenameRecoveryFailsForDirFsLevel() throws Exception {
     if (getConfiguration().getIsClientTransactionIdEnabled()) {
       // Recovery based on client transaction id should be successful
       assertTrue(renameResult);
-      // One extra getPathStatus call should have happened
-      newConnections = 5;
     } else {
       assertFalse(renameResult);
-      newConnections = 4;
     }
 
     // validating stat counters after rename
-    // 3 calls should have happened in total for rename
+    // 4 calls should have happened in total for rename
     // 1 -> original rename rest call, 2 -> first retry,
     // +1 for getPathStatus calls
     // last getPathStatus call should be skipped
     assertThatStatisticCounter(ioStats,
             CONNECTIONS_MADE.getStatName())
-            .isEqualTo(newConnections + connMadeBeforeRename);
+            .isEqualTo(4 + connMadeBeforeRename);
 
     // the RENAME_PATH_ATTEMPTS stat should be incremented by 1
     // retries happen internally within AbfsRestOperation execute()