diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
index 5cf7ec565b..e1b791f6ef 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/AbfsHttpConstants.java
@@ -111,6 +111,17 @@ public final class AbfsHttpConstants {
   public static final char CHAR_EQUALS = '=';
   public static final char CHAR_STAR = '*';
   public static final char CHAR_PLUS = '+';
+  /**
+   * Value that differentiates categories of the http_status.
+   * <pre>
+   * 100 - 199 : Informational responses
+   * 200 - 299 : Successful responses
+   * 300 - 399 : Redirection messages
+   * 400 - 499 : Client error responses
+   * 500 - 599 : Server error responses
+   * </pre>
+   */
+  public static final Integer HTTP_STATUS_CATEGORY_QUOTIENT = 100;
 
   private AbfsHttpConstants() {}
 }
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java
index 8bc31c4f92..8a5e9db855 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java
@@ -66,6 +66,10 @@ public String getErrorCode() {
     return this.errorCode;
   }
 
+  public String getErrorMessage() {
+    return this.errorMessage;
+  }
+
   public static List<AzureServiceErrorCode> getAzureServiceCode(int httpStatusCode) {
     List<AzureServiceErrorCode> errorCodes = new ArrayList<>();
     if (httpStatusCode == UNKNOWN.httpStatusCode) {
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java
index 00da9b6601..ad99020390 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java
@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.fs.azurebfs.AbfsStatistic;
 import org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
@@ -73,6 +74,12 @@ public class AbfsRestOperation {
   private AbfsHttpOperation result;
   private AbfsCounters abfsCounters;
 
+  /**
+   * This variable contains the reason of last API call within the same
+   * AbfsRestOperation object.
+   */
+  private String failureReason;
+
   /**
    * Checks if there is non-null HTTP response.
    * @return true if there is a non-null HTTP response from the ABFS call.
@@ -208,7 +215,7 @@ public void execute(TracingContext tracingContext)
   private void completeExecute(TracingContext tracingContext)
       throws AzureBlobFileSystemException {
     // see if we have latency reports from the previous requests
-    String latencyHeader = this.client.getAbfsPerfTracker().getClientLatency();
+    String latencyHeader = getClientLatency();
     if (latencyHeader != null && !latencyHeader.isEmpty()) {
       AbfsHttpHeader httpHeader =
               new AbfsHttpHeader(HttpHeaderConfigurations.X_MS_ABFS_CLIENT_LATENCY, latencyHeader);
@@ -237,6 +244,11 @@ private void completeExecute(TracingContext tracingContext)
     LOG.trace("{} REST operation complete", operationType);
   }
 
+  @VisibleForTesting
+  String getClientLatency() {
+    return client.getAbfsPerfTracker().getClientLatency();
+  }
+
   /**
    * Executes a single HTTP operation to complete the REST operation.  If it
    * fails, there may be a retry.  The retryCount is incremented with each
@@ -248,9 +260,9 @@ private boolean executeHttpOperation(final int retryCount,
 
     try {
       // initialize the HTTP request and open the connection
-      httpOperation = new AbfsHttpOperation(url, method, requestHeaders);
+      httpOperation = createHttpOperation();
       incrementCounter(AbfsStatistic.CONNECTIONS_MADE, 1);
-      tracingContext.constructHeader(httpOperation);
+      tracingContext.constructHeader(httpOperation, failureReason);
 
       switch(client.getAuthType()) {
         case Custom:
@@ -303,6 +315,7 @@ private boolean executeHttpOperation(final int retryCount,
     } catch (UnknownHostException ex) {
       String hostname = null;
       hostname = httpOperation.getHost();
+      failureReason = RetryReason.getAbbreviation(ex, null, null);
       LOG.warn("Unknown host name: {}. Retrying to resolve the host name...",
           hostname);
       if (!client.getRetryPolicy().shouldRetry(retryCount, -1)) {
@@ -314,6 +327,8 @@ private boolean executeHttpOperation(final int retryCount,
         LOG.debug("HttpRequestFailure: {}, {}", httpOperation, ex);
       }
 
+      failureReason = RetryReason.getAbbreviation(ex, -1, "");
+
       if (!client.getRetryPolicy().shouldRetry(retryCount, -1)) {
         throw new InvalidAbfsRestOperationException(ex);
       }
@@ -326,6 +341,8 @@ private boolean executeHttpOperation(final int retryCount,
     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;
     }
 
@@ -334,6 +351,15 @@ private boolean executeHttpOperation(final int retryCount,
     return true;
   }
 
+  /**
+   * Creates new object of {@link AbfsHttpOperation} with the url, method, and
+   * requestHeaders fields of the AbfsRestOperation object.
+   */
+  @VisibleForTesting
+  AbfsHttpOperation createHttpOperation() throws IOException {
+    return new AbfsHttpOperation(url, method, requestHeaders);
+  }
+
   /**
    * Incrementing Abfs counters with a long value.
    *
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReason.java
new file mode 100644
index 0000000000..40e8cdc1e0
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReason.java
@@ -0,0 +1,102 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.ClientErrorRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.ConnectionResetRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.ConnectionTimeoutRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.ReadTimeoutRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.RetryReasonCategory;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.ServerErrorRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.UnknownHostRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.UnknownIOExceptionRetryReason;
+import org.apache.hadoop.fs.azurebfs.services.retryReasonCategories.UnknownSocketExceptionRetryReason;
+
+
+/**
+ * This utility class exposes methods to convert a server response-error to a
+ * category of error.
+ */
+final class RetryReason {
+
+  /**
+   * Linked-list of the implementations of RetryReasonCategory. The objects in the
+   * list are arranged by the rank of their significance.
+   * <ul>
+   *   <li>ServerError (statusCode==5XX), ClientError (statusCode==4XX) are
+   *   independent of other retryReason categories.</li>
+   *   <li>Since {@link java.net.SocketException} is subclass of
+   *   {@link java.io.IOException},
+   *   hence, {@link UnknownIOExceptionRetryReason} is placed before
+   *   {@link UnknownSocketExceptionRetryReason}</li>
+   *   <li>Since, connectionTimeout, readTimeout, and connectionReset are
+   *   {@link java.net.SocketTimeoutException} exceptions with different messages,
+   *   hence, {@link ConnectionTimeoutRetryReason}, {@link ReadTimeoutRetryReason},
+   *   {@link ConnectionResetRetryReason} are above {@link UnknownIOExceptionRetryReason}.
+   *   There is no order between the three reasons as they are differentiated
+   *   by exception-message.</li>
+   *   <li>Since, {@link java.net.UnknownHostException} is subclass of
+   *   {@link java.io.IOException}, {@link UnknownHostRetryReason} is placed
+   *   over {@link UnknownIOExceptionRetryReason}</li>
+   * </ul>
+   */
+  private static List<RetryReasonCategory> rankedReasonCategories
+      = new LinkedList<RetryReasonCategory>() {{
+    add(new ServerErrorRetryReason());
+    add(new ClientErrorRetryReason());
+    add(new UnknownIOExceptionRetryReason());
+    add(new UnknownSocketExceptionRetryReason());
+    add(new ConnectionTimeoutRetryReason());
+    add(new ReadTimeoutRetryReason());
+    add(new UnknownHostRetryReason());
+    add(new ConnectionResetRetryReason());
+  }};
+
+  private RetryReason() {
+
+  }
+
+  /**
+   * Method to get correct abbreviation for a given set of exception, statusCode,
+   * storageStatusCode.
+   *
+   * @param ex exception caught during server communication.
+   * @param statusCode statusCode in the server response.
+   * @param storageErrorMessage storageErrorMessage in the server response.
+   *
+   * @return abbreviation for the the given set of exception, statusCode, storageStatusCode.
+   */
+  static String getAbbreviation(Exception ex,
+      Integer statusCode,
+      String storageErrorMessage) {
+    String result = null;
+    for (RetryReasonCategory retryReasonCategory : rankedReasonCategories) {
+      final String abbreviation
+          = retryReasonCategory.captureAndGetAbbreviation(ex,
+          statusCode, storageErrorMessage);
+      if (abbreviation != null) {
+        result = abbreviation;
+      }
+    }
+    return result;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReasonConstants.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReasonConstants.java
new file mode 100644
index 0000000000..8a0af183e3
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/RetryReasonConstants.java
@@ -0,0 +1,39 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+public final class RetryReasonConstants {
+
+  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 UNKNOWN_HOST_EXCEPTION_ABBREVIATION = "UH";
+  public static final String IO_EXCEPTION_ABBREVIATION = "IOE";
+  public static final String SOCKET_EXCEPTION_ABBREVIATION = "SE";
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ClientErrorRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ClientErrorRetryReason.java
new file mode 100644
index 0000000000..cf1c47e3eb
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ClientErrorRetryReason.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_STATUS_CATEGORY_QUOTIENT;
+
+/**
+ * Category that can capture server-response errors for 4XX status-code.
+ */
+public class ClientErrorRetryReason extends RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    if (statusCode == null || statusCode / HTTP_STATUS_CATEGORY_QUOTIENT != 4) {
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return statusCode + "";
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionResetRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionResetRetryReason.java
new file mode 100644
index 0000000000..702f887564
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionResetRetryReason.java
@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_RESET_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_RESET_MESSAGE;
+
+/**
+ * Category that can capture server-response errors for connection-reset exception.
+ */
+public class ConnectionResetRetryReason extends
+    RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    return checkExceptionMessage(ex, CONNECTION_RESET_MESSAGE);
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return CONNECTION_RESET_ABBREVIATION;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionTimeoutRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionTimeoutRetryReason.java
new file mode 100644
index 0000000000..28f35dcc80
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ConnectionTimeoutRetryReason.java
@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_JDK_MESSAGE;
+
+/**
+ * Category that can capture server-response errors for connection-timeout.
+ */
+public class ConnectionTimeoutRetryReason extends
+    RetryReasonCategory {
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return CONNECTION_TIMEOUT_ABBREVIATION;
+  }
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    return checkExceptionMessage(ex, CONNECTION_TIMEOUT_JDK_MESSAGE);
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ReadTimeoutRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ReadTimeoutRetryReason.java
new file mode 100644
index 0000000000..4663d9a52b
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ReadTimeoutRetryReason.java
@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.READ_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.READ_TIMEOUT_JDK_MESSAGE;
+
+/**
+ * Category that can capture server-response errors for read-timeout.
+ */
+public class ReadTimeoutRetryReason extends RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    return checkExceptionMessage(ex, READ_TIMEOUT_JDK_MESSAGE);
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return READ_TIMEOUT_ABBREVIATION;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/RetryReasonCategory.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/RetryReasonCategory.java
new file mode 100644
index 0000000000..893451b496
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/RetryReasonCategory.java
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import java.util.Locale;
+
+/**
+ * Provides methods to define if given exception can be categorised to certain category.
+ * Each category has a different implementation of the abstract class.
+ */
+public abstract class RetryReasonCategory {
+
+  /**
+   * Returns if given server response error can be categorised by the implementation.
+   *
+   * @param ex exception captured in the server response.
+   * @param statusCode statusCode on the server response
+   * @param serverErrorMessage serverErrorMessage on the server response.
+   *
+   * @return <ol><li>true if server response error can be categorised by the implementation</li>
+   * <li>false if response error can not be categorised by the implementation</li></ol>
+   */
+  abstract Boolean canCapture(Exception ex,
+      Integer statusCode,
+      String serverErrorMessage);
+
+  /**
+   * Returns the abbreviation corresponding to the server response error.
+   *
+   * @param statusCode statusCode on the server response
+   * @param serverErrorMessage serverErrorMessage on the server response.
+   *
+   * @return abbreviation on the basis of the statusCode and the serverErrorMessage
+   */
+  abstract String getAbbreviation(Integer statusCode, String serverErrorMessage);
+
+  /**
+   * Converts the server-error response to an abbreviation if the response can be
+   * categorised by the implementation.
+   *
+   * @param ex exception received while making API request
+   * @param statusCode statusCode received in the server-response
+   * @param serverErrorMessage error-message received in the server-response
+   *
+   * @return abbreviation if the server-response can be categorised by the implementation.
+   * null if the server-response can not be categorised by the implementation.
+   */
+  public String captureAndGetAbbreviation(Exception ex,
+      Integer statusCode,
+      String serverErrorMessage) {
+    if (canCapture(ex, statusCode, serverErrorMessage)) {
+      return getAbbreviation(statusCode, serverErrorMessage);
+    }
+    return null;
+  }
+
+  /**
+   * Checks if a required search-string is in the exception's message.
+   */
+  Boolean checkExceptionMessage(final Exception exceptionCaptured,
+      final String search) {
+    if (search == null) {
+      return false;
+    }
+    if (exceptionCaptured != null
+        && exceptionCaptured.getMessage() != null
+        && exceptionCaptured.getMessage()
+        .toLowerCase(Locale.US)
+        .contains(search.toLowerCase(Locale.US))) {
+      return true;
+    }
+    return false;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ServerErrorRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ServerErrorRetryReason.java
new file mode 100644
index 0000000000..dd67a0cb8c
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/ServerErrorRetryReason.java
@@ -0,0 +1,67 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+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;
+
+/**
+ * Category that can capture server-response errors for 5XX status-code.
+ */
+public class ServerErrorRetryReason extends RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    if (statusCode == null || statusCode / HTTP_STATUS_CATEGORY_QUOTIENT != 5) {
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    if (statusCode == HTTP_UNAVAILABLE && serverErrorMessage != null) {
+      String splitedServerErrorMessage = serverErrorMessage.split(System.lineSeparator(),
+          2)[0];
+      if (INGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage().equalsIgnoreCase(
+          splitedServerErrorMessage)) {
+        return INGRESS_LIMIT_BREACH_ABBREVIATION;
+      }
+      if (EGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage().equalsIgnoreCase(
+          splitedServerErrorMessage)) {
+        return EGRESS_LIMIT_BREACH_ABBREVIATION;
+      }
+      if (OPERATION_BREACH_MESSAGE.equalsIgnoreCase(
+          splitedServerErrorMessage)) {
+        return OPERATION_LIMIT_BREACH_ABBREVIATION;
+      }
+      return HTTP_UNAVAILABLE + "";
+    }
+    return statusCode + "";
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownHostRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownHostRetryReason.java
new file mode 100644
index 0000000000..c329348d81
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownHostRetryReason.java
@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import java.net.UnknownHostException;
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.UNKNOWN_HOST_EXCEPTION_ABBREVIATION;
+
+/**
+ * Category that can capture server-response errors for {@link UnknownHostException}.
+ */
+public class UnknownHostRetryReason extends RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    if (ex instanceof UnknownHostException) {
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return UNKNOWN_HOST_EXCEPTION_ABBREVIATION;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownIOExceptionRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownIOExceptionRetryReason.java
new file mode 100644
index 0000000000..8a69ebb928
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownIOExceptionRetryReason.java
@@ -0,0 +1,47 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.IO_EXCEPTION_ABBREVIATION;
+
+
+/**
+ * Category that can capture server-response errors for {@link IOException}.
+ */
+public class UnknownIOExceptionRetryReason extends
+    RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    if (ex instanceof IOException) {
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return IO_EXCEPTION_ABBREVIATION;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownSocketExceptionRetryReason.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownSocketExceptionRetryReason.java
new file mode 100644
index 0000000000..18e9f115fe
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/UnknownSocketExceptionRetryReason.java
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import java.net.SocketException;
+
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.SOCKET_EXCEPTION_ABBREVIATION;
+
+/**
+ * Category that can capture server-response errors for {@link SocketException}.
+ */
+public class UnknownSocketExceptionRetryReason extends
+    RetryReasonCategory {
+
+  @Override
+  Boolean canCapture(final Exception ex,
+      final Integer statusCode,
+      final String serverErrorMessage) {
+    if (ex instanceof SocketException) {
+      return true;
+    }
+    return false;
+  }
+
+  @Override
+  String getAbbreviation(final Integer statusCode,
+      final String serverErrorMessage) {
+    return SOCKET_EXCEPTION_ABBREVIATION;
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/package-info.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/package-info.java
new file mode 100644
index 0000000000..7d8078620a
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/retryReasonCategories/package-info.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * A retryReasonCategory defines methods applicable on server-response errors.
+ */
+@Private
+@Evolving
+package org.apache.hadoop.fs.azurebfs.services.retryReasonCategories;
+
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Evolving;
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/TracingContext.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/TracingContext.java
index 5a115451df..9a2ccda36f 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/TracingContext.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/TracingContext.java
@@ -152,8 +152,10 @@ public void setListener(Listener listener) {
    * X_MS_CLIENT_REQUEST_ID header of the http operation
    * @param httpOperation AbfsHttpOperation instance to set header into
    *                      connection
+   * @param previousFailure List of failures seen before this API trigger on
+   * same operation from AbfsClient.
    */
-  public void constructHeader(AbfsHttpOperation httpOperation) {
+  public void constructHeader(AbfsHttpOperation httpOperation, String previousFailure) {
     clientRequestId = UUID.randomUUID().toString();
     switch (format) {
     case ALL_ID_FORMAT: // Optional IDs (e.g. streamId) may be empty
@@ -161,6 +163,7 @@ public void constructHeader(AbfsHttpOperation httpOperation) {
           clientCorrelationID + ":" + clientRequestId + ":" + fileSystemID + ":"
               + primaryRequestId + ":" + streamID + ":" + opType + ":"
               + retryCount;
+      header = addFailureReasons(header, previousFailure);
       break;
     case TWO_ID_FORMAT:
       header = clientCorrelationID + ":" + clientRequestId;
@@ -174,6 +177,14 @@ public void constructHeader(AbfsHttpOperation httpOperation) {
     httpOperation.setRequestProperty(HttpHeaderConfigurations.X_MS_CLIENT_REQUEST_ID, header);
   }
 
+  private String addFailureReasons(final String header,
+      final String previousFailure) {
+    if (previousFailure == null) {
+      return header;
+    }
+    return String.format("%s_%s", header, previousFailure);
+  }
+
   /**
    * Return header representing the request associated with the tracingContext
    * @return Header string set into X_MS_CLIENT_REQUEST_ID
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsRestOperationMockFailures.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsRestOperationMockFailures.java
new file mode 100644
index 0000000000..bfa524a25e
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsRestOperationMockFailures.java
@@ -0,0 +1,302 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.net.HttpURLConnection;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.stubbing.Stubber;
+
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+
+import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
+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.AuthType.OAuth;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_RESET_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_RESET_MESSAGE;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_JDK_MESSAGE;
+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.IO_EXCEPTION_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.services.RetryReasonConstants.READ_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.READ_TIMEOUT_JDK_MESSAGE;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.SOCKET_EXCEPTION_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.UNKNOWN_HOST_EXCEPTION_ABBREVIATION;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.nullable;
+
+public class TestAbfsRestOperationMockFailures {
+
+  @Test
+  public void testClientRequestIdForConnectTimeoutRetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new SocketTimeoutException(CONNECTION_TIMEOUT_JDK_MESSAGE);
+    abbreviations[0] = CONNECTION_TIMEOUT_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForConnectAndReadTimeoutRetry()
+      throws Exception {
+    Exception[] exceptions = new Exception[2];
+    String[] abbreviations = new String[2];
+    exceptions[0] = new SocketTimeoutException(CONNECTION_TIMEOUT_JDK_MESSAGE);
+    abbreviations[0] = CONNECTION_TIMEOUT_ABBREVIATION;
+    exceptions[1] = new SocketTimeoutException(READ_TIMEOUT_JDK_MESSAGE);
+    abbreviations[1] = READ_TIMEOUT_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForReadTimeoutRetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new SocketTimeoutException(READ_TIMEOUT_JDK_MESSAGE);
+    abbreviations[0] = READ_TIMEOUT_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForUnknownHostRetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new UnknownHostException();
+    abbreviations[0] = UNKNOWN_HOST_EXCEPTION_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForConnectionResetRetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new SocketTimeoutException(CONNECTION_RESET_MESSAGE + " by peer");
+    abbreviations[0] = CONNECTION_RESET_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForUnknownSocketExRetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new SocketException("unknown");
+    abbreviations[0] = SOCKET_EXCEPTION_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdForIOERetry() throws Exception {
+    Exception[] exceptions = new Exception[1];
+    String[] abbreviations = new String[1];
+    exceptions[0] = new InterruptedIOException();
+    abbreviations[0] = IO_EXCEPTION_ABBREVIATION;
+    testClientRequestIdForTimeoutRetry(exceptions, abbreviations, 1);
+  }
+
+  @Test
+  public void testClientRequestIdFor400Retry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_BAD_REQUEST, "", "400");
+  }
+
+  @Test
+  public void testClientRequestIdFor500Retry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_INTERNAL_ERROR, "", "500");
+  }
+
+  @Test
+  public void testClientRequestIdFor503INGRetry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_UNAVAILABLE,
+        INGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage(),
+        INGRESS_LIMIT_BREACH_ABBREVIATION);
+  }
+
+  @Test
+  public void testClientRequestIdFor503egrRetry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_UNAVAILABLE,
+        EGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage(),
+        EGRESS_LIMIT_BREACH_ABBREVIATION);
+  }
+
+  @Test
+  public void testClientRequestIdFor503OPRRetry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_UNAVAILABLE,
+        OPERATION_BREACH_MESSAGE, OPERATION_LIMIT_BREACH_ABBREVIATION);
+  }
+
+  @Test
+  public void testClientRequestIdFor503OtherRetry() throws Exception {
+    testClientRequestIdForStatusRetry(HTTP_UNAVAILABLE, "Other.", "503");
+  }
+
+  private void testClientRequestIdForStatusRetry(int status,
+      String serverErrorMessage,
+      String keyExpected) throws Exception {
+
+    AbfsClient abfsClient = Mockito.mock(AbfsClient.class);
+    ExponentialRetryPolicy retryPolicy = Mockito.mock(
+        ExponentialRetryPolicy.class);
+    addMockBehaviourToAbfsClient(abfsClient, retryPolicy);
+
+
+    AbfsRestOperation abfsRestOperation = Mockito.spy(new AbfsRestOperation(
+        AbfsRestOperationType.ReadFile,
+        abfsClient,
+        "PUT",
+        null,
+        new ArrayList<>()
+    ));
+
+    AbfsHttpOperation httpOperation = Mockito.mock(AbfsHttpOperation.class);
+    addMockBehaviourToRestOpAndHttpOp(abfsRestOperation, httpOperation);
+
+    Mockito.doNothing()
+        .doNothing()
+        .when(httpOperation)
+        .processResponse(nullable(byte[].class), nullable(int.class),
+            nullable(int.class));
+
+    int[] statusCount = new int[1];
+    statusCount[0] = 0;
+    Mockito.doAnswer(answer -> {
+      if (statusCount[0] <= 5) {
+        statusCount[0]++;
+        return status;
+      }
+      return HTTP_OK;
+    }).when(httpOperation).getStatusCode();
+
+    Mockito.doReturn(serverErrorMessage)
+        .when(httpOperation)
+        .getStorageErrorMessage();
+
+    TracingContext tracingContext = Mockito.mock(TracingContext.class);
+    Mockito.doNothing().when(tracingContext).setRetryCount(nullable(int.class));
+
+    int[] count = new int[1];
+    count[0] = 0;
+    Mockito.doAnswer(invocationOnMock -> {
+      if (count[0] == 1) {
+        Assertions.assertThat((String) invocationOnMock.getArgument(1))
+            .isEqualTo(keyExpected);
+      }
+      count[0]++;
+      return null;
+    }).when(tracingContext).constructHeader(any(), any());
+
+    abfsRestOperation.execute(tracingContext);
+    Assertions.assertThat(count[0]).isEqualTo(2);
+
+  }
+
+  private void testClientRequestIdForTimeoutRetry(Exception[] exceptions,
+      String[] abbreviationsExpected,
+      int len) throws Exception {
+    AbfsClient abfsClient = Mockito.mock(AbfsClient.class);
+    ExponentialRetryPolicy retryPolicy = Mockito.mock(
+        ExponentialRetryPolicy.class);
+    addMockBehaviourToAbfsClient(abfsClient, retryPolicy);
+
+
+    AbfsRestOperation abfsRestOperation = Mockito.spy(new AbfsRestOperation(
+        AbfsRestOperationType.ReadFile,
+        abfsClient,
+        "PUT",
+        null,
+        new ArrayList<>()
+    ));
+
+    AbfsHttpOperation httpOperation = Mockito.mock(AbfsHttpOperation.class);
+    addMockBehaviourToRestOpAndHttpOp(abfsRestOperation, httpOperation);
+
+    Stubber stubber = Mockito.doThrow(exceptions[0]);
+    for (int iteration = 1; iteration < len; iteration++) {
+      stubber.doThrow(exceptions[iteration]);
+    }
+    stubber
+        .doNothing()
+        .when(httpOperation)
+        .processResponse(nullable(byte[].class), nullable(int.class),
+            nullable(int.class));
+
+    Mockito.doReturn(HTTP_OK).when(httpOperation).getStatusCode();
+
+    TracingContext tracingContext = Mockito.mock(TracingContext.class);
+    Mockito.doNothing().when(tracingContext).setRetryCount(nullable(int.class));
+
+    int[] count = new int[1];
+    count[0] = 0;
+    Mockito.doAnswer(invocationOnMock -> {
+      if (count[0] > 0 && count[0] <= len) {
+        Assertions.assertThat((String) invocationOnMock.getArgument(1))
+            .isEqualTo(abbreviationsExpected[count[0] - 1]);
+      }
+      count[0]++;
+      return null;
+    }).when(tracingContext).constructHeader(any(), any());
+
+    abfsRestOperation.execute(tracingContext);
+    Assertions.assertThat(count[0]).isEqualTo(len + 1);
+  }
+
+  private void addMockBehaviourToRestOpAndHttpOp(final AbfsRestOperation abfsRestOperation,
+      final AbfsHttpOperation httpOperation) throws IOException {
+    HttpURLConnection httpURLConnection = Mockito.mock(HttpURLConnection.class);
+    Mockito.doNothing()
+        .when(httpURLConnection)
+        .setRequestProperty(nullable(String.class), nullable(String.class));
+    Mockito.doReturn(httpURLConnection).when(httpOperation).getConnection();
+    Mockito.doReturn("").when(abfsRestOperation).getClientLatency();
+    Mockito.doReturn(httpOperation).when(abfsRestOperation).createHttpOperation();
+  }
+
+  private void addMockBehaviourToAbfsClient(final AbfsClient abfsClient,
+      final ExponentialRetryPolicy retryPolicy) 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)
+        .sendingRequest(any(), nullable(AbfsCounters.class));
+    Mockito.doNothing().when(intercept).updateMetrics(any(), any());
+
+    Mockito.doReturn(retryPolicy).when(abfsClient).getRetryPolicy();
+    Mockito.doReturn(true)
+        .when(retryPolicy)
+        .shouldRetry(nullable(Integer.class), nullable(Integer.class));
+    Mockito.doReturn(false).when(retryPolicy).shouldRetry(1, HTTP_OK);
+    Mockito.doReturn(false).when(retryPolicy).shouldRetry(2, HTTP_OK);
+  }
+}
diff --git a/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestRetryReason.java b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestRetryReason.java
new file mode 100644
index 0000000000..76fcc6dc2c
--- /dev/null
+++ b/hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestRetryReason.java
@@ -0,0 +1,134 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.io.IOException;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.net.UnknownHostException;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import static java.net.HttpURLConnection.HTTP_FORBIDDEN;
+import static java.net.HttpURLConnection.HTTP_INTERNAL_ERROR;
+import static java.net.HttpURLConnection.HTTP_UNAVAILABLE;
+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.CONNECTION_RESET_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_RESET_MESSAGE;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.CONNECTION_TIMEOUT_JDK_MESSAGE;
+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.IO_EXCEPTION_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.services.RetryReasonConstants.READ_TIMEOUT_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.READ_TIMEOUT_JDK_MESSAGE;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.SOCKET_EXCEPTION_ABBREVIATION;
+import static org.apache.hadoop.fs.azurebfs.services.RetryReasonConstants.UNKNOWN_HOST_EXCEPTION_ABBREVIATION;
+
+public class TestRetryReason {
+
+  @Test
+  public void test4xxStatusRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_FORBIDDEN, null))
+        .describedAs("Abbreviation for 4xx should be equal to 4xx")
+        .isEqualTo(HTTP_FORBIDDEN + "");
+  }
+
+  @Test
+  public void testConnectionResetRetryReason() {
+    SocketException connReset = new SocketException(CONNECTION_RESET_MESSAGE.toUpperCase());
+    Assertions.assertThat(RetryReason.getAbbreviation(connReset, null, null)).isEqualTo(CONNECTION_RESET_ABBREVIATION);
+  }
+
+  @Test
+  public void testConnectionTimeoutRetryReason() {
+    SocketTimeoutException connectionTimeoutException = new SocketTimeoutException(CONNECTION_TIMEOUT_JDK_MESSAGE);
+    Assertions.assertThat(RetryReason.getAbbreviation(connectionTimeoutException, null, null)).isEqualTo(
+        CONNECTION_TIMEOUT_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testReadTimeoutRetryReason() {
+    SocketTimeoutException connectionTimeoutException = new SocketTimeoutException(READ_TIMEOUT_JDK_MESSAGE);
+    Assertions.assertThat(RetryReason.getAbbreviation(connectionTimeoutException, null, null)).isEqualTo(
+        READ_TIMEOUT_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testEgressLimitRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_UNAVAILABLE, EGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage())).isEqualTo(
+        EGRESS_LIMIT_BREACH_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testIngressLimitRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_UNAVAILABLE, INGRESS_OVER_ACCOUNT_LIMIT.getErrorMessage())).isEqualTo(
+        INGRESS_LIMIT_BREACH_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testOperationLimitRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_UNAVAILABLE, OPERATION_BREACH_MESSAGE)).isEqualTo(
+        OPERATION_LIMIT_BREACH_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void test503UnknownRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_UNAVAILABLE, null)).isEqualTo(
+        "503"
+    );
+  }
+
+  @Test
+  public void test500RetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(null, HTTP_INTERNAL_ERROR, null)).isEqualTo(
+        "500"
+    );
+  }
+
+  @Test
+  public void testUnknownHostRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(new UnknownHostException(), null, null)).isEqualTo(
+        UNKNOWN_HOST_EXCEPTION_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testUnknownIOExceptionRetryReason() {
+    Assertions.assertThat(RetryReason.getAbbreviation(new IOException(), null, null)).isEqualTo(
+        IO_EXCEPTION_ABBREVIATION
+    );
+  }
+
+  @Test
+  public void testUnknownSocketException() {
+    Assertions.assertThat(RetryReason.getAbbreviation(new SocketException(), null, null)).isEqualTo(
+        SOCKET_EXCEPTION_ABBREVIATION
+    );
+  }
+}