HADOOP-19096: [ABFS] [CST Optimization] Enhancing Client-Side Throttling Metrics Updating Logic

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

@@ -42,8 +42,14 @@ public enum AzureServiceErrorCode {
   INVALID_SOURCE_OR_DESTINATION_RESOURCE_TYPE("InvalidSourceOrDestinationResourceType", HttpURLConnection.HTTP_CONFLICT, null),
   RENAME_DESTINATION_PARENT_PATH_NOT_FOUND("RenameDestinationParentPathNotFound", HttpURLConnection.HTTP_NOT_FOUND, null),
   INVALID_RENAME_SOURCE_PATH("InvalidRenameSourcePath", HttpURLConnection.HTTP_CONFLICT, null),
-  INGRESS_OVER_ACCOUNT_LIMIT(null, HttpURLConnection.HTTP_UNAVAILABLE, "Ingress is over the account limit."),
-  EGRESS_OVER_ACCOUNT_LIMIT(null, HttpURLConnection.HTTP_UNAVAILABLE, "Egress is over the account limit."),
+  INGRESS_OVER_ACCOUNT_LIMIT("ServerBusy", HttpURLConnection.HTTP_UNAVAILABLE,
+          "Ingress is over the account limit."),
+  EGRESS_OVER_ACCOUNT_LIMIT("ServerBusy", HttpURLConnection.HTTP_UNAVAILABLE,
+          "Egress is over the account limit."),
+  TPS_OVER_ACCOUNT_LIMIT("ServerBusy", HttpURLConnection.HTTP_UNAVAILABLE,
+          "Operations per second is over the account limit."),
+  OTHER_SERVER_THROTTLING("ServerBusy", HttpURLConnection.HTTP_UNAVAILABLE,
+          "The server is currently unable to receive requests. Please retry your request."),
   INVALID_QUERY_PARAMETER_VALUE("InvalidQueryParameterValue", HttpURLConnection.HTTP_BAD_REQUEST, null),
   AUTHORIZATION_PERMISSION_MISS_MATCH("AuthorizationPermissionMismatch", HttpURLConnection.HTTP_FORBIDDEN, null),
   ACCOUNT_REQUIRES_HTTPS("AccountRequiresHttps", HttpURLConnection.HTTP_BAD_REQUEST, null),

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

@@ -39,6 +39,7 @@
 import org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding;
 
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_CONTINUE;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.*;
 
 /**
  * The AbfsRestOperation for Rest AbfsClient.
@@ -277,6 +278,7 @@ String getClientLatency() {
   private boolean executeHttpOperation(final int retryCount,
     TracingContext tracingContext) throws AzureBlobFileSystemException {
     AbfsHttpOperation httpOperation;
+    boolean wasExceptionThrown = false;
 
     try {
       // initialize the HTTP request and open the connection
@@ -314,7 +316,23 @@ private boolean executeHttpOperation(final int retryCount,
       } else if (httpOperation.getStatusCode() == HttpURLConnection.HTTP_UNAVAILABLE) {
         incrementCounter(AbfsStatistic.SERVER_UNAVAILABLE, 1);
       }
+
+      // If no exception occurred here it means http operation was successfully complete and
+      // a response from server has been received which might be failure or success.
+      LOG.debug("HttpRequest: {}: {}", operationType, httpOperation);
+
+      // If request failed at server and should be retried, we will determine failure reason and retry policy here
+      if (client.getRetryPolicy().shouldRetry(retryCount, httpOperation.getStatusCode())) {
+        int status = httpOperation.getStatusCode();
+        failureReason = RetryReason.getAbbreviation(null, status, httpOperation.getStorageErrorMessage());
+        return false;
+      }
+
+      // If the request has succeeded or failed with non-retrial error, save the operation and return.
+      result = httpOperation;
+
     } catch (UnknownHostException ex) {
+      wasExceptionThrown = true;
       String hostname = null;
       hostname = httpOperation.getHost();
       failureReason = RetryReason.getAbbreviation(ex, null, null);
@@ -325,6 +343,7 @@ private boolean executeHttpOperation(final int retryCount,
       }
       return false;
     } catch (IOException ex) {
+      wasExceptionThrown = true;
       if (LOG.isDebugEnabled()) {
         LOG.debug("HttpRequestFailure: {}, {}", httpOperation, ex);
       }
@@ -337,32 +356,31 @@ private boolean executeHttpOperation(final int retryCount,
 
       return false;
     } finally {
-      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 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
-        throttling backoff computation.
+        Updating Client Side Throttling Metrics for relevant response status codes.
+        1. Status code in 2xx range: Successful Operations should contribute
+        2. Status code in 3xx range: Redirection Operations should not contribute
+        3. Status code in 4xx range: User Errors should not contribute
+        4. Status code in 503 range: Server Error should contribute as following:
+          a. 503, Ingress Over Account Limit: Should Contribute
+          b. 503, Egress Over Account Limit: Should Contribute
+          c. 503, TPS Over Account Limit: Should Contribute
+          d. 503, Other Server Throttling: Should not contribute
+        5. Status code in 5xx range other than 503: Should not contribute
+        6. IOException and UnknownHostExceptions: Should not contribute
        */
-      boolean updateMetricsResponseCode = (status < HttpURLConnection.HTTP_MULT_CHOICE
-              || status >= HttpURLConnection.HTTP_INTERNAL_ERROR);
-      if (updateMetricsResponseCode) {
+      int statusCode = httpOperation.getStatusCode();
+      boolean shouldUpdateCSTMetrics = (statusCode <  HttpURLConnection.HTTP_MULT_CHOICE // Case 1
+              || INGRESS_LIMIT_BREACH_ABBREVIATION.equals(failureReason) // Case 4.a
+              || EGRESS_LIMIT_BREACH_ABBREVIATION.equals(failureReason) // Case 4.b
+              || OPERATION_LIMIT_BREACH_ABBREVIATION.equals(failureReason)) // Case 4.c
+              && !wasExceptionThrown; // Case 6
+
+      if (shouldUpdateCSTMetrics) {
         intercept.updateMetrics(operationType, httpOperation);
       }
     }
 
-    LOG.debug("HttpRequest: {}: {}", operationType, httpOperation);
-
-    if (client.getRetryPolicy().shouldRetry(retryCount, httpOperation.getStatusCode())) {
-      int status = httpOperation.getStatusCode();
-      failureReason = RetryReason.getAbbreviation(null, status, httpOperation.getStorageErrorMessage());
-      return false;
-    }
-
-    result = httpOperation;
-
     return true;
   }
 

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

@@ -26,13 +26,13 @@ private RetryReasonConstants() {
   public static final String CONNECTION_TIMEOUT_JDK_MESSAGE = "connect timed out";
   public static final String READ_TIMEOUT_JDK_MESSAGE = "Read timed out";
   public static final String CONNECTION_RESET_MESSAGE = "Connection reset";
-  public static final String OPERATION_BREACH_MESSAGE = "Operations per second is over the account limit.";
   public static final String CONNECTION_RESET_ABBREVIATION = "CR";
   public static final String CONNECTION_TIMEOUT_ABBREVIATION = "CT";
   public static final String READ_TIMEOUT_ABBREVIATION = "RT";
   public static final String INGRESS_LIMIT_BREACH_ABBREVIATION = "ING";
   public static final String EGRESS_LIMIT_BREACH_ABBREVIATION = "EGR";
   public static final String OPERATION_LIMIT_BREACH_ABBREVIATION = "OPR";
+  public static final String OTHER_SERVER_THROTTLING_ABBREVIATION = "OTH";
   public static final String UNKNOWN_HOST_EXCEPTION_ABBREVIATION = "UH";
   public static final String IO_EXCEPTION_ABBREVIATION = "IOE";
   public static final String SOCKET_EXCEPTION_ABBREVIATION = "SE";

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

@@ -20,12 +20,8 @@
 
 import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_STATUS_CATEGORY_QUOTIENT;
-import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.EGRESS_OVER_ACCOUNT_LIMIT;
-import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.INGRESS_OVER_ACCOUNT_LIMIT;
-import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.EGRESS_LIMIT_BREACH_ABBREVIATION;
-import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.INGRESS_LIMIT_BREACH_ABBREVIATION;
-import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.OPERATION_BREACH_MESSAGE;
-import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.OPERATION_LIMIT_BREACH_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode.*;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.*;
 
 /**
  * Category that can capture server-response errors for 5XX status-code.
@@ -56,10 +52,14 @@ String getAbbreviation(final Integer statusCode,
           splitedServerErrorMessage)) {
         return EGRESS_LIMIT_BREACH_ABBREVIATION;
       }
-      if (OPERATION_BREACH_MESSAGE.equalsIgnoreCase(
+      if (TPS_OVER_ACCOUNT_LIMIT.getErrorMessage().equalsIgnoreCase(
           splitedServerErrorMessage)) {
         return OPERATION_LIMIT_BREACH_ABBREVIATION;
       }
+      if (OTHER_SERVER_THROTTLING.getErrorMessage().equalsIgnoreCase(
+          splitedServerErrorMessage)) {
+        return OTHER_SERVER_THROTTLING_ABBREVIATION;
+      }
       return HTTP_UNAVAILABLE + "";
     }
     return statusCode + "";

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

@@ -56,6 +56,7 @@ public static void setMockAbfsRestOperationForListPathOperation(
       FunctionRaisingIOE<AbfsHttpOperation, AbfsHttpOperation> functionRaisingIOE)
       throws Exception {
     ExponentialRetryPolicy retryPolicy = Mockito.mock(ExponentialRetryPolicy.class);
+    AbfsThrottlingIntercept intercept = Mockito.mock(AbfsThrottlingIntercept.class);
     AbfsHttpOperation httpOperation = Mockito.mock(AbfsHttpOperation.class);
     AbfsRestOperation abfsRestOperation = Mockito.spy(new AbfsRestOperation(
         AbfsRestOperationType.ListPaths,
@@ -68,7 +69,7 @@ public static void setMockAbfsRestOperationForListPathOperation(
     Mockito.doReturn(abfsRestOperation).when(spiedClient).getAbfsRestOperation(
         eq(AbfsRestOperationType.ListPaths), any(), any(), any());
 
-    addGeneralMockBehaviourToAbfsClient(spiedClient, retryPolicy);
+    addGeneralMockBehaviourToAbfsClient(spiedClient, retryPolicy, intercept);
     addGeneralMockBehaviourToRestOpAndHttpOp(abfsRestOperation, httpOperation);
 
     functionRaisingIOE.apply(httpOperation);
@@ -100,11 +101,10 @@ public static void addGeneralMockBehaviourToRestOpAndHttpOp(final AbfsRestOperat
    * @throws IOException
    */
   public static void addGeneralMockBehaviourToAbfsClient(final AbfsClient abfsClient,
-                                                         final ExponentialRetryPolicy retryPolicy) throws IOException {
+                                                         final ExponentialRetryPolicy retryPolicy,
+                                                         final AbfsThrottlingIntercept intercept) throws IOException {
     Mockito.doReturn(OAuth).when(abfsClient).getAuthType();
     Mockito.doReturn("").when(abfsClient).getAccessToken();
-    AbfsThrottlingIntercept intercept = Mockito.mock(
-        AbfsThrottlingIntercept.class);
     Mockito.doReturn(intercept).when(abfsClient).getIntercept();
     Mockito.doNothing()
         .when(intercept)