HADOOP-18146: ABFS: Added changes for expect hundred continue header

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

@@ -64,6 +64,11 @@ public final class AbfsHttpConstants {
   public static final String HTTP_METHOD_PATCH = "PATCH";
   public static final String HTTP_METHOD_POST = "POST";
   public static final String HTTP_METHOD_PUT = "PUT";
+  /**
+   * All status codes less than http 100 signify error
+   * and should qualify for retry.
+   */
+  public static final int HTTP_CONTINUE = 100;
 
   // Abfs generic constants
   public static final String SINGLE_WHITE_SPACE = " ";

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

@@ -35,6 +35,10 @@ public final class ConfigurationKeys {
    * path to determine HNS status.
    */
   public static final String FS_AZURE_ACCOUNT_IS_HNS_ENABLED = "fs.azure.account.hns.enabled";
+  /**
+   * Enable or disable expect hundred continue header.
+   * Value: {@value}.
+   */
   public static final String FS_AZURE_ACCOUNT_IS_EXPECT_HEADER_ENABLED = "fs.azure.account.expect.header.enabled";
   public static final String FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME = "fs.azure.account.key";
   public static final String FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME_REGX = "fs\\.azure\\.account\\.key\\.(.*)";

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

@@ -686,8 +686,13 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
 
     final URL url = createRequestUrl(path, abfsUriQueryBuilder.toString());
     final AbfsRestOperation op = getAbfsRestOperationForAppend(AbfsRestOperationType.Append,
-        HTTP_METHOD_PUT, url, requestHeaders, buffer, reqParams.getoffset(),
-        reqParams.getLength(), sasTokenForReuse);
+            HTTP_METHOD_PUT,
+            url,
+            requestHeaders,
+            buffer,
+            reqParams.getoffset(),
+            reqParams.getLength(),
+            sasTokenForReuse);
     try {
       op.execute(tracingContext);
     } catch (AzureBlobFileSystemException e) {
@@ -702,7 +707,7 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
       */
       int responseStatusCode = ((AbfsRestOperationException) e).getStatusCode();
       if (checkUserError(responseStatusCode) && reqParams.isExpectHeaderEnabled()) {
-        LOG.debug("User error, retrying without 100 continue enabled");
+        LOG.debug("User error, retrying without 100 continue enabled for the given path " + path);
         reqParams.setExpectHeaderEnabled(false);
         return this.append(path, buffer, reqParams, cachedSasToken,
             tracingContext);
@@ -715,9 +720,14 @@ public AbfsRestOperation append(final String path, final byte[] buffer,
           && appendSuccessCheckOp(op, path,
           (reqParams.getPosition() + reqParams.getLength()), tracingContext)) {
         final AbfsRestOperation successOp = getAbfsRestOperationForAppend(
-            AbfsRestOperationType.Append, HTTP_METHOD_PUT, url, requestHeaders,
-            buffer, reqParams.getoffset(), reqParams.getLength(),
-            sasTokenForReuse);
+                AbfsRestOperationType.Append,
+                HTTP_METHOD_PUT,
+                url,
+                requestHeaders,
+                buffer,
+                reqParams.getoffset(),
+                reqParams.getLength(),
+                sasTokenForReuse);
         successOp.hardSetResult(HttpURLConnection.HTTP_OK);
         return successOp;
       }
@@ -728,7 +738,7 @@ && appendSuccessCheckOp(op, path,
   }
 
   /**
-   * Returns the rest operation for append
+   * Returns the rest operation for append.
    * @param operationType The AbfsRestOperationType.
    * @param httpMethod specifies the httpMethod.
    * @param url specifies the url.
@@ -737,7 +747,7 @@ && appendSuccessCheckOp(op, path,
    * @param bufferOffset The buffer offset.
    * @param bufferLength The buffer Length.
    * @param sasTokenForReuse The sasToken.
-   * @return AbfsRestOperation op
+   * @return AbfsRestOperation op.
    */
   @VisibleForTesting
   AbfsRestOperation getAbfsRestOperationForAppend(final AbfsRestOperationType operationType,

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

@@ -62,7 +62,7 @@ public AbfsClientThrottlingIntercept(String accountName, AbfsConfiguration abfsC
 
   // Hide default constructor
   private AbfsClientThrottlingIntercept(AbfsConfiguration abfsConfiguration) {
-    //Account name is kept as empty as same instance is shared across all accounts.
+    // Account name is kept as empty as same instance is shared across all accounts.
     this.accountName = "";
     this.readThrottler = setAnalyzer("read", abfsConfiguration);
     this.writeThrottler = setAnalyzer("write", abfsConfiguration);
@@ -117,11 +117,11 @@ static AbfsClientThrottlingIntercept initializeSingleton(AbfsConfiguration abfsC
   }
 
   /**
-   * Updates the metrics for the case when getOutputStream() caught an IOException
-   * and response code signifies throttling.
+   * Updates the metrics for the case when response code signifies throttling
+   * but there are some expected bytes to be sent.
    * @param isThrottledOperation returns true if status code is HTTP_UNAVAILABLE
    * @param abfsHttpOperation Used for status code and data transferred.
-   * @return
+   * @return true if the operation is throttled and has some bytes to transfer.
    */
   private boolean updateBytesTransferred(boolean isThrottledOperation,
       AbfsHttpOperation abfsHttpOperation) {
@@ -148,7 +148,7 @@ public void updateMetrics(AbfsRestOperationType operationType,
     boolean isFailedOperation = (status < HttpURLConnection.HTTP_OK
         || status >= HttpURLConnection.HTTP_INTERNAL_ERROR);
 
-    // If status code is 503, it considered a throttled operation.
+    // If status code is 503, it is considered as a throttled operation.
     boolean isThrottledOperation = (status == HTTP_UNAVAILABLE);
 
     switch (operationType) {
@@ -160,7 +160,7 @@ public void updateMetrics(AbfsRestOperationType operationType,
             throttling but there were some expectedBytesToBeSent.
            */
           if (updateBytesTransferred(isThrottledOperation, abfsHttpOperation)) {
-            LOG.debug("Updating metrics due to throttling");
+            LOG.debug("Updating metrics due to throttling for path " + abfsHttpOperation.getConnUrl().getPath());
             contentLength = abfsHttpOperation.getExpectedBytesToBeSent();
           }
         }

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

@@ -37,6 +37,8 @@
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_CONTINUE;
+
 /**
  * The AbfsRestOperation for Rest AbfsClient.
  */
@@ -61,9 +63,6 @@ public class AbfsRestOperation {
   // Used only by AbfsInputStream/AbfsOutputStream to reuse SAS tokens.
   private final String sasToken;
 
-  // All status codes less than http 100 signify error.
-  private static final int HTTP_CONTINUE = 100;
-
   private static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);
 
   // For uploads, this is the request entity body.  For downloads,
@@ -250,6 +249,14 @@ private void completeExecute(TracingContext tracingContext)
     LOG.trace("{} REST operation complete", operationType);
   }
 
+  /**
+   * Returns a new object of AbfsHttpOperation.
+   * @param url The url for the operation.
+   * @param method The http method.
+   * @param requestHeaders The request headers for the operation.
+   * @return AbfsHttpOperation object.
+   * @throws IOException
+   */
   AbfsHttpOperation getHttpOperation(final URL url, final String method,
       final List<AbfsHttpHeader> requestHeaders) throws IOException {
     return new AbfsHttpOperation(url, method, requestHeaders);
@@ -341,7 +348,7 @@ private boolean executeHttpOperation(final int retryCount,
       int status = httpOperation.getStatusCode();
       /*
         A status less than 300 (2xx range) or greater than or equal
-        to 500 (5xx range) should contribute to throttling metrics updation.
+        to 500 (5xx range) should contribute to throttling metrics being updated.
         Less than 200 or greater than or equal to 500 show failed operations. 2xx
         range contributes to successful operations. 3xx range is for redirects
         and 4xx range is for user errors. These should not be a part of

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

@@ -24,6 +24,8 @@
 import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
 import org.apache.hadoop.classification.VisibleForTesting;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_CONTINUE;
+
 /**
  * Retry policy used by AbfsClient.
  * */
@@ -56,12 +58,6 @@ public class ExponentialRetryPolicy {
    */
   private static final double MAX_RANDOM_RATIO = 1.2;
 
-  /**
-   * All status codes less than http 100 signify error
-   * and should qualify for retry.
-   */
-  private static final int HTTP_CONTINUE = 100;
-
   /**
    *  Holds the random number generator used to calculate randomized backoff intervals
    */

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

@@ -42,7 +42,7 @@
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsOutputStream;
 import org.apache.hadoop.fs.azurebfs.services.AuthType;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
 import org.apache.hadoop.fs.azure.AzureNativeFileSystemStore;
 import org.apache.hadoop.fs.azure.NativeAzureFileSystem;
 import org.apache.hadoop.fs.azure.metrics.AzureFileSystemInstrumentation;
@@ -254,7 +254,7 @@ public Hashtable<String, String> call() throws Exception {
   }
 
   public AccessTokenProvider getAccessTokenProvider(final AzureBlobFileSystem fs) {
-    return TestAbfsClient.getAccessTokenProvider(fs.getAbfsStore().getClient());
+    return ITestAbfsClient.getAccessTokenProvider(fs.getAbfsStore().getClient());
   }
 
   public void loadConfiguredFileSystem() throws Exception {

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

@@ -43,7 +43,7 @@
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
 import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
 
@@ -362,7 +362,7 @@ public void testNegativeScenariosForCreateOverwriteDisabled()
     // Get mock AbfsClient with current config
     AbfsClient
         mockClient
-        = TestAbfsClient.getMockAbfsClient(
+        = ITestAbfsClient.getMockAbfsClient(
         fs.getAbfsStore().getClient(),
         fs.getAbfsStore().getAbfsConfiguration());
 

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

@@ -35,7 +35,7 @@
 import org.apache.hadoop.fs.azurebfs.services.AbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.AbfsHttpOperation;
 import org.apache.hadoop.fs.azurebfs.services.AbfsRestOperation;
-import org.apache.hadoop.fs.azurebfs.services.TestAbfsClient;
+import org.apache.hadoop.fs.azurebfs.services.ITestAbfsClient;
 import org.apache.hadoop.fs.azurebfs.services.TestAbfsPerfTracker;
 import org.apache.hadoop.fs.azurebfs.utils.TestMockHelpers;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
@@ -176,7 +176,7 @@ public void testDeleteIdempotency() throws Exception {
 
     final AzureBlobFileSystem fs = getFileSystem();
     AbfsClient abfsClient = fs.getAbfsStore().getClient();
-    AbfsClient testClient = TestAbfsClient.createTestClientFromCurrentContext(
+    AbfsClient testClient = ITestAbfsClient.createTestClientFromCurrentContext(
         abfsClient,
         abfsConfig);
 
@@ -223,7 +223,7 @@ public void testDeleteIdempotency() throws Exception {
   public void testDeleteIdempotencyTriggerHttp404() throws Exception {
 
     final AzureBlobFileSystem fs = getFileSystem();
-    AbfsClient client = TestAbfsClient.createTestClientFromCurrentContext(
+    AbfsClient client = ITestAbfsClient.createTestClientFromCurrentContext(
         fs.getAbfsStore().getClient(),
         this.getConfiguration());
 
@@ -242,7 +242,7 @@ public void testDeleteIdempotencyTriggerHttp404() throws Exception {
         getTestTracingContext(fs, true)));
 
     // mock idempotency check to mimic retried case
-    AbfsClient mockClient = TestAbfsClient.getMockAbfsClient(
+    AbfsClient mockClient = ITestAbfsClient.getMockAbfsClient(
         fs.getAbfsStore().getClient(),
         this.getConfiguration());
     AzureBlobFileSystemStore mockStore = mock(AzureBlobFileSystemStore.class);
@@ -257,10 +257,10 @@ public void testDeleteIdempotencyTriggerHttp404() throws Exception {
 
     // Case 2: Mimic retried case
     // Idempotency check on Delete always returns success
-    AbfsRestOperation idempotencyRetOp = TestAbfsClient.getRestOp(
+    AbfsRestOperation idempotencyRetOp = ITestAbfsClient.getRestOp(
         DeletePath, mockClient, HTTP_METHOD_DELETE,
-        TestAbfsClient.getTestUrl(mockClient, "/NonExistingPath"),
-        TestAbfsClient.getTestRequestHeaders(mockClient));
+        ITestAbfsClient.getTestUrl(mockClient, "/NonExistingPath"),
+        ITestAbfsClient.getTestRequestHeaders(mockClient));
     idempotencyRetOp.hardSetResult(HTTP_OK);
 
     doReturn(idempotencyRetOp).when(mockClient).deleteIdempotencyCheckOp(any());

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

@@ -82,7 +82,7 @@
  * Test useragent of abfs client.
  *
  */
-public final class TestAbfsClient extends AbstractAbfsIntegrationTest {
+public final class ITestAbfsClient extends AbstractAbfsIntegrationTest {
 
   private static final String ACCOUNT_NAME = "bogusAccountName.dfs.core.windows.net";
   private static final String FS_AZURE_USER_AGENT_PREFIX = "Partner Service";
@@ -94,7 +94,7 @@ public final class TestAbfsClient extends AbstractAbfsIntegrationTest {
 
   private final Pattern userAgentStringPattern;
 
-  public TestAbfsClient() throws Exception {
+  public ITestAbfsClient() throws Exception {
     StringBuilder regEx = new StringBuilder();
     regEx.append("^");
     regEx.append(APN_VERSION);
@@ -340,30 +340,30 @@ public static AbfsClient getMockAbfsClient(AbfsClient baseAbfsClientInstance,
             abfsConfig.getAccountName().substring(0,
                 abfsConfig.getAccountName().indexOf(DOT)), abfsConfig));
     // override baseurl
-    client = TestAbfsClient.setAbfsClientField(client, "abfsConfiguration",
+    client = ITestAbfsClient.setAbfsClientField(client, "abfsConfiguration",
         abfsConfig);
 
     // override baseurl
-    client = TestAbfsClient.setAbfsClientField(client, "baseUrl",
+    client = ITestAbfsClient.setAbfsClientField(client, "baseUrl",
         baseAbfsClientInstance.getBaseUrl());
 
     // override auth provider
     if (currentAuthType == AuthType.SharedKey) {
-      client = TestAbfsClient.setAbfsClientField(client, "sharedKeyCredentials",
+      client = ITestAbfsClient.setAbfsClientField(client, "sharedKeyCredentials",
           new SharedKeyCredentials(
               abfsConfig.getAccountName().substring(0,
                   abfsConfig.getAccountName().indexOf(DOT)),
               abfsConfig.getStorageAccountKey()));
     } else {
-      client = TestAbfsClient.setAbfsClientField(client, "tokenProvider",
+      client = ITestAbfsClient.setAbfsClientField(client, "tokenProvider",
           abfsConfig.getTokenProvider());
     }
 
     // override user agent
     String userAgent = "APN/1.0 Azure Blob FS/3.4.0-SNAPSHOT (PrivateBuild "
         + "JavaJRE 1.8.0_252; Linux 5.3.0-59-generic/amd64; openssl-1.0; "
         + "UNKNOWN/UNKNOWN) MSFT";
-    client = TestAbfsClient.setAbfsClientField(client, "userAgent", userAgent);
+    client = ITestAbfsClient.setAbfsClientField(client, "userAgent", userAgent);
 
     return client;
   }
@@ -469,7 +469,7 @@ public void testExpectHundredContinue() throws Exception {
 
     // Gets the client.
     AbfsClient testClient = Mockito.spy(
-        TestAbfsClient.createTestClientFromCurrentContext(
+        ITestAbfsClient.createTestClientFromCurrentContext(
             abfsClient,
             abfsConfig));
 
@@ -489,7 +489,7 @@ public void testExpectHundredContinue() throws Exception {
 
     // Creates a list of request headers.
     final List<AbfsHttpHeader> requestHeaders
-        = TestAbfsClient.getTestRequestHeaders(testClient);
+        = ITestAbfsClient.getTestRequestHeaders(testClient);
     requestHeaders.add(
         new AbfsHttpHeader(X_HTTP_METHOD_OVERRIDE, HTTP_METHOD_PATCH));
     if (appendRequestParameters.isExpectHeaderEnabled()) {
@@ -566,9 +566,7 @@ public void testExpectHundredContinue() throws Exception {
         TracingHeaderFormat.ALL_ID_FORMAT, null));
 
     // Check that expect header is enabled before the append call.
-    Assertions.assertThat(appendRequestParameters.isExpectHeaderEnabled())
-        .describedAs("The expect header is not true before the append call")
-            .isEqualTo(true);
+    assertTrue(appendRequestParameters.isExpectHeaderEnabled());
 
     intercept(AzureBlobFileSystemException.class,
         () -> testClient.append(finalTestPath, buffer, appendRequestParameters, null, tracingContext));
@@ -579,8 +577,6 @@ public void testExpectHundredContinue() throws Exception {
         .isEqualTo(0);
 
     // Verify that the same request was retried with expect header disabled.
-    Assertions.assertThat(appendRequestParameters.isExpectHeaderEnabled())
-        .describedAs("The expect header is not false")
-        .isEqualTo(false);
+    assertFalse(appendRequestParameters.isExpectHeaderEnabled());
   }
 }