Code Improvements

Author: anujmodi2021

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/exceptions/TailLatencyRequestTimeoutException.java

@@ -18,15 +18,18 @@
 
 package org.apache.hadoop.fs.azurebfs.contracts.exceptions;
 
+import java.util.concurrent.TimeoutException;
 import static org.apache.hadoop.fs.azurebfs.services.AbfsErrors.ERR_TAIL_LATENCY_REQUEST_TIMEOUT;
 
+/**
+ * Thrown when a request takes more time than the current reported tail latency.
+ */
 public class TailLatencyRequestTimeoutException extends AzureBlobFileSystemException {
 
-  public TailLatencyRequestTimeoutException() {
-    super(ERR_TAIL_LATENCY_REQUEST_TIMEOUT);
-  }
-
-  public TailLatencyRequestTimeoutException(Exception innerException) {
+  /**
+   * Constructs a TailLatencyRequestTimeoutException with TimeoutException as the cause.
+   */
+  public TailLatencyRequestTimeoutException(TimeoutException innerException) {
     super(ERR_TAIL_LATENCY_REQUEST_TIMEOUT, innerException);
   }
 }

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

@@ -91,14 +91,19 @@ public class AbfsAHCHttpOperation extends AbfsHttpOperation {
    */
   private final AbfsApacheHttpClient abfsApacheHttpClient;
 
-  private final long tailLatencyMs;
+  /**
+   * Timeout in milliseconds that defines maximum allowed time to execute operation.
+   * This timeout starts when execution starts and includes E2E processing time of request.
+   * This is based on tail latency observed in the system.
+   */
+  private final long tailLatencyTimeout;
 
   public AbfsAHCHttpOperation(final URL url,
       final String method,
       final List<AbfsHttpHeader> requestHeaders,
       final Duration connectionTimeout,
       final Duration readTimeout,
-      final long tailLatencyMs,
+      final long tailLatencyTimeout,
       final AbfsApacheHttpClient abfsApacheHttpClient,
       final AbfsClient abfsClient) throws IOException {
     super(LOG, url, method, requestHeaders, connectionTimeout, readTimeout,
@@ -107,7 +112,7 @@ public AbfsAHCHttpOperation(final URL url,
         || HTTP_METHOD_PATCH.equals(method)
         || HTTP_METHOD_POST.equals(method);
     this.abfsApacheHttpClient = abfsApacheHttpClient;
-    this.tailLatencyMs = tailLatencyMs;
+    this.tailLatencyTimeout = tailLatencyTimeout;
     LOG.debug("Creating AbfsAHCHttpOperation for URL: {}, method: {}",
         url, method);
 
@@ -163,8 +168,8 @@ AbfsManagedHttpClientContext getHttpClientContext() {
     return new AbfsManagedHttpClientContext();
   }
 
-  long getTailLatencyMs() {
-    return tailLatencyMs;
+  long getTailLatencyTimeout() {
+    return tailLatencyTimeout;
   }
 
   /**{@inheritDoc}*/
@@ -281,7 +286,7 @@ HttpResponse executeRequest() throws IOException {
     try {
       LOG.debug("Executing request: {}", httpRequestBase);
       HttpResponse response = abfsApacheHttpClient.execute(httpRequestBase,
-          abfsHttpClientContext, getConnectionTimeout(), getReadTimeout(), getTailLatencyMs());
+          abfsHttpClientContext, getConnectionTimeout(), getReadTimeout(), getTailLatencyTimeout());
       setConnectionTimeMs(abfsHttpClientContext.getConnectTime());
       setSendRequestTimeMs(abfsHttpClientContext.getSendTime());
       setRecvResponseTimeMs(abfsHttpClientContext.getReadTime());

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

@@ -141,13 +141,36 @@ public HttpResponse execute(HttpRequestBase httpRequest,
       final AbfsManagedHttpClientContext abfsHttpClientContext,
       final int connectTimeout,
       final int readTimeout,
-      final long tailLatency) throws IOException {
-    if (tailLatency <= 0) {
-      return executeWithDeadline(httpRequest, abfsHttpClientContext,
-          connectTimeout, readTimeout, Long.MAX_VALUE);
+      final long tailLatencyTimeout) throws IOException {
+    if (tailLatencyTimeout <= 0) {
+      return executeWithoutDeadline(httpRequest, abfsHttpClientContext,
+          connectTimeout, readTimeout);
     }
     return executeWithDeadline(httpRequest, abfsHttpClientContext,
-        connectTimeout, readTimeout, tailLatency);
+        connectTimeout, readTimeout, tailLatencyTimeout);
+  }
+
+  /**
+   * Executes the HTTP request.
+   *
+   * @param httpRequest HTTP request to execute.
+   * @param abfsHttpClientContext HttpClient context.
+   * @param connectTimeout Connection timeout.
+   * @param readTimeout Read timeout.
+   *
+   * @return HTTP response.
+   * @throws IOException network error.
+   */
+  public HttpResponse executeWithoutDeadline(HttpRequestBase httpRequest,
+      final AbfsManagedHttpClientContext abfsHttpClientContext,
+      final int connectTimeout,
+      final int readTimeout) throws IOException {
+    RequestConfig.Builder requestConfigBuilder = RequestConfig
+        .custom()
+        .setConnectTimeout(connectTimeout)
+        .setSocketTimeout(readTimeout);
+    httpRequest.setConfig(requestConfigBuilder.build());
+    return httpClient.execute(httpRequest, abfsHttpClientContext);
   }
 
   /**
@@ -168,22 +191,27 @@ public HttpResponse executeWithDeadline(HttpRequestBase httpRequest,
       final int connectTimeout,
       final int readTimeout,
       final long deadlineMillis) throws IOException {
-
     RequestConfig.Builder requestConfigBuilder = RequestConfig
         .custom()
         .setConnectTimeout(connectTimeout)
         .setSocketTimeout(readTimeout);
     httpRequest.setConfig(requestConfigBuilder.build());
-
     ExecutorService executor = Executors.newSingleThreadExecutor();
     Future<HttpResponse> future = executor.submit(() ->
-        httpClient.execute(httpRequest, abfsHttpClientContext));
+        httpClient.execute(httpRequest, abfsHttpClientContext)
+    );
+
     try {
       return future.get(deadlineMillis, TimeUnit.MILLISECONDS);
     } catch (TimeoutException e) {
+      /* Deadline exceeded, abort the request.
+       * This will also kill the underlying socket exception in the HttpClient.
+       * Connection will be marker stale and won't be returned back to KAC for reuse.
+       */
       httpRequest.abort();
       throw new TailLatencyRequestTimeoutException(e);
     } catch (Exception e) {
+      // Any other exception from execution should be thrown as IOException.
       throw new IOException("Request execution with deadline failed", e);
     } finally {
       executor.shutdownNow();

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

@@ -190,7 +190,7 @@ public abstract class AbfsClient implements Closeable {
   private EncryptionContextProvider encryptionContextProvider = null;
   private EncryptionType encryptionType = EncryptionType.NONE;
   private final AbfsThrottlingIntercept intercept;
-  private AbfsTailLatencyTracker latencyTracker = null;
+  private AbfsTailLatencyTracker tailLatencyTracker = null;
 
   private final ListeningScheduledExecutorService executorService;
 
@@ -227,7 +227,7 @@ private AbfsClient(final URL baseUrl,
     this.accountName = abfsConfiguration.getAccountName().substring(0, abfsConfiguration.getAccountName().indexOf(AbfsHttpConstants.DOT));
     this.authType = abfsConfiguration.getAuthType(accountName);
     this.intercept = AbfsThrottlingInterceptFactory.getInstance(accountName, abfsConfiguration);
-    this.latencyTracker = AbfsTailLatencyTrackerFactory.getInstance(accountName, abfsConfiguration);
+    this.tailLatencyTracker = AbfsTailLatencyTrackerFactory.getInstance(accountName, abfsConfiguration);
     this.renameResilience = abfsConfiguration.getRenameResilience();
     this.abfsServiceType = abfsServiceType;
 
@@ -434,8 +434,12 @@ public void setEncryptionType(EncryptionType encryptionType) {
     this.encryptionType = encryptionType;
   }
 
-  public AbfsTailLatencyTracker getLatencyTracker() {
-    return latencyTracker;
+  /**
+   * Get the tail latency tracker for this account.
+   * @return tail latency tracker if enabled, null otherwise.
+   */
+  public AbfsTailLatencyTracker getTailLatencyTracker() {
+    return tailLatencyTracker;
   }
 
   public EncryptionType getEncryptionType() {

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

@@ -80,7 +80,7 @@ public class AbfsRestOperation {
   private final AbfsClient client;
   // Return intercept instance
   private final AbfsThrottlingIntercept intercept;
-  // Latency tracker
+  // Tail Latency tracker
   private final AbfsTailLatencyTracker tailLatencyTracker;
   // the HTTP method (PUT, PATCH, POST, GET, HEAD, or DELETE)
   private final String method;
@@ -236,7 +236,7 @@ String getSasToken() {
     }
     this.maxIoRetries = abfsConfiguration.getMaxIoRetries();
     this.intercept = client.getIntercept();
-    this.tailLatencyTracker = client.getLatencyTracker();
+    this.tailLatencyTracker = client.getTailLatencyTracker();
     this.abfsConfiguration = abfsConfiguration;
     this.retryPolicy = client.getExponentialRetryPolicy();
   }
@@ -524,7 +524,12 @@ private boolean executeHttpOperation(final int retryCount,
         }
       }
       if (!retryPolicy.shouldRetry(retryCount, -1)) {
+        /*
+         * If a request is failing with TailLatencyTimeout exception.
+         * it should be retried without Tail Latency Timeout exception should not be returned to caller.
+         */
         if (retryPolicy instanceof TailLatencyRequestTimeoutRetryPolicy) {
+          // Disable Tail Latency Timeout for the next retry.
           shouldTailLatencyTimeout = false;
           return false;
         }
@@ -540,6 +545,7 @@ private boolean executeHttpOperation(final int retryCount,
         intercept.updateMetrics(operationType, httpOperation);
       }
 
+      // Update Tail Latency Tracker only for successful requests.
       if (tailLatencyTracker != null && statusCode <  HttpURLConnection.HTTP_MULT_CHOICE) {
         tailLatencyTracker.updateLatency(operationType,
             httpOperation.getSendLatency() + httpOperation.getRecvLatency());
@@ -624,14 +630,19 @@ AbfsJdkHttpOperation createAbfsHttpOperation() throws IOException {
 
   @VisibleForTesting
   AbfsAHCHttpOperation createAbfsAHCHttpOperation() throws IOException {
-    long tailLatency = getTailLatencyIfTimeoutEnabled();
+    long tailLatency = getTailLatencyTimeoutIfEnabled();
     return new AbfsAHCHttpOperation(url, method, requestHeaders,
         Duration.ofMillis(client.getAbfsConfiguration().getHttpConnectionTimeout()),
         Duration.ofMillis(client.getAbfsConfiguration().getHttpReadTimeout()),
         tailLatency, client.getAbfsApacheHttpClient(), client);
   }
 
-  long getTailLatencyIfTimeoutEnabled() {
+  /**
+   * Get Tail Latency Timeout value if profiling is enabled, timeout is enabled
+   * and retries due to tail latency request timeout is allowed.
+   * @return tail latency timeout value else return zero.
+   */
+  long getTailLatencyTimeoutIfEnabled() {
     if (tailLatencyTracker != null && abfsConfiguration.isTailLatencyRequestTimeoutEnabled() && shouldTailLatencyTimeout) {
       return (long) tailLatencyTracker.getTailLatency(this.operationType);
     }

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

@@ -144,8 +144,9 @@ public void updateLatency(final AbfsRestOperationType operationType,
   }
 
   /**
-   *
+   * Gets the tail latency for a specific operation type.
    * @param operationType Only applicable for read and write operations.
+   * @return Tail latency value.
    */
   public double getTailLatency(final AbfsRestOperationType operationType) {
     SlidingWindowHdrHistogram histogram = operationLatencyMap.get(operationType);

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

@@ -1,3 +1,21 @@
+/**
+ * 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.AbfsErrors.ERR_TAIL_LATENCY_REQUEST_TIMEOUT;
@@ -18,4 +36,4 @@ Boolean canCapture(final Exception ex,
       final String serverErrorMessage) {
     return checkExceptionMessage(ex, ERR_TAIL_LATENCY_REQUEST_TIMEOUT);
   }
-}
+}