Re: [PR] HADOOP-18325: ABFS: Add correlated metric support for ABFS operations [hadoop]

["steveloughran (via GitHub)" <gi...@apache.org>]

steveloughran commented on code in PR #6314:
URL: https://github.com/apache/hadoop/pull/6314#discussion_r1575011941


##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsCounters.java:
##########
@@ -19,16 +19,17 @@
 package org.apache.hadoop.fs.azurebfs.services;
 
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
 
 import org.apache.hadoop.classification.VisibleForTesting;
-
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.azurebfs.AbfsStatistic;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
 import org.apache.hadoop.fs.statistics.IOStatisticsSource;
-
+import org.apache.hadoop.fs.azurebfs.AbfsBackoffMetrics;

Review Comment:
   import ordering



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -1731,6 +1792,128 @@ protected AccessTokenProvider getTokenProvider() {
     return tokenProvider;
   }
 
+  /**
+   * Retrieves a TracingContext object configured for metric tracking.
+   * This method creates a TracingContext object with the validated client correlation ID,
+   * the host name of the local machine (or "UnknownHost" if unable to determine),
+   * the file system operation type set to GET_ATTR, and additional configuration parameters
+   * for metric tracking.
+   * The TracingContext is intended for use in tracking metrics related to Azure Blob FileSystem (ABFS) operations.
+   *
+   * @return A TracingContext object configured for metric tracking.
+   */
+  private TracingContext getMetricTracingContext() {
+    String hostName;
+    try {
+      hostName = InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      hostName = "UnknownHost";
+    }
+    return new TracingContext(TracingContext.validateClientCorrelationID(
+        abfsConfiguration.getClientCorrelationId()),
+        hostName, FSOperationType.GET_ATTR, true,
+        abfsConfiguration.getTracingHeaderFormat(),
+        null, abfsCounters.toString());
+  }
+
+  /**
+   * Synchronized method to suspend or resume timer.
+   * @param timerFunctionality resume or suspend.
+   * @param timerTask The timertask object.
+   * @return true or false.
+   */
+  boolean timerOrchestrator(TimerFunctionality timerFunctionality, TimerTask timerTask) {
+    switch (timerFunctionality) {
+      case RESUME:
+        if (isMetricCollectionStopped.get()) {
+          synchronized (this) {
+            if (isMetricCollectionStopped.get()) {
+              resumeTimer();
+            }
+          }
+        }
+        break;
+      case SUSPEND:
+        long now = System.currentTimeMillis();
+        long lastExecutionTime = abfsCounters.getLastExecutionTime().get();
+        if (isMetricCollectionEnabled && (now - lastExecutionTime >= metricAnalysisPeriod)) {
+          synchronized (this) {
+            if (!isMetricCollectionStopped.get()) {
+              timerTask.cancel();
+              timer.purge();
+              isMetricCollectionStopped.set(true);
+              return true;
+            }
+          }
+        }
+        break;
+      default:
+        break;
+    }
+    return false;
+  }
+
+  private void resumeTimer() {

Review Comment:
   if this is only used on L1831 it may be best to inline: guarantees where it is used. alternatively, make synchronized and move the `if (isMetricCollectionStopped.get()` clause in too



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsCountersImpl.java:
##########
@@ -33,9 +33,12 @@
 import org.apache.hadoop.metrics2.lib.MetricsRegistry;
 import org.apache.hadoop.metrics2.lib.MutableCounterLong;
 import org.apache.hadoop.metrics2.lib.MutableMetric;
-
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.*;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.iostatisticsStore;
+import static org.apache.hadoop.util.Time.now;
+
+import java.util.concurrent.atomic.AtomicLong;

Review Comment:
   move up to the java. import block. 



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java:
##########
@@ -105,24 +109,22 @@ public static AzureServiceErrorCode getAzureServiceCode(int httpStatusCode, Stri
         return azureServiceErrorCode;
       }
     }
-
     return UNKNOWN;
   }
 
   public static AzureServiceErrorCode getAzureServiceCode(int httpStatusCode, String errorCode, final String errorMessage) {

Review Comment:
   this is complex enough it absolutely needs javadocs



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsCountersImpl.java:
##########
@@ -33,9 +33,12 @@
 import org.apache.hadoop.metrics2.lib.MetricsRegistry;
 import org.apache.hadoop.metrics2.lib.MutableCounterLong;
 import org.apache.hadoop.metrics2.lib.MutableMetric;
-
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;

Review Comment:
   nit: move to L27



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/MetricFormat.java:
##########
@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs.utils;
+
+public enum MetricFormat {

Review Comment:
   1. javadocs
   2. dangerously generic name. 



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap
+      = new ConcurrentHashMap<>();
+
+  public AbfsBackoffMetrics() {
+    initializeMap();
+    this.numberOfIOPSThrottledRequests = new AtomicLong();
+    this.numberOfBandwidthThrottledRequests = new AtomicLong();
+    this.numberOfOtherThrottledRequests = new AtomicLong();
+    this.totalNumberOfRequests = new AtomicLong();
+    this.maxRetryCount = new AtomicLong();
+    this.numberOfRequestsSucceededWithoutRetrying = new AtomicLong();
+    this.numberOfRequestsFailed = new AtomicLong();
+    this.numberOfNetworkFailedRequests = new AtomicLong();
+  }
+
+  public AbfsBackoffMetrics(String retryCount) {

Review Comment:
   this only partially initialises the metrics



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap
+      = new ConcurrentHashMap<>();
+
+  public AbfsBackoffMetrics() {
+    initializeMap();
+    this.numberOfIOPSThrottledRequests = new AtomicLong();
+    this.numberOfBandwidthThrottledRequests = new AtomicLong();
+    this.numberOfOtherThrottledRequests = new AtomicLong();
+    this.totalNumberOfRequests = new AtomicLong();
+    this.maxRetryCount = new AtomicLong();
+    this.numberOfRequestsSucceededWithoutRetrying = new AtomicLong();
+    this.numberOfRequestsFailed = new AtomicLong();
+    this.numberOfNetworkFailedRequests = new AtomicLong();
+  }
+
+  public AbfsBackoffMetrics(String retryCount) {
+    this.retryCount = retryCount;
+    this.numberOfRequestsSucceeded = new AtomicLong();
+    this.minBackoff = new AtomicLong(Long.MAX_VALUE);
+    this.maxBackoff = new AtomicLong();
+    this.totalRequests = new AtomicLong();
+    this.totalBackoff = new AtomicLong();
+  }
+
+  private void initializeMap() {
+    ArrayList<String> retryCountList = new ArrayList<String>(
+        Arrays.asList("1", "2", "3", "4", "5_15", "15_25", "25AndAbove"));
+    for (String s : retryCountList) {
+      metricsMap.put(s, new AbfsBackoffMetrics(s));
+    }
+  }
+
+  public long getNumberOfRequestsSucceeded() {
+    return this.numberOfRequestsSucceeded.get();
+  }
+
+  public void setNumberOfRequestsSucceeded(long numberOfRequestsSucceeded) {
+    this.numberOfRequestsSucceeded.set(numberOfRequestsSucceeded);
+  }
+
+  public void incrementNumberOfRequestsSucceeded() {
+    this.numberOfRequestsSucceeded.getAndIncrement();
+  }
+
+  public long getMinBackoff() {
+    return this.minBackoff.get();
+  }
+
+  public void setMinBackoff(long minBackoff) {
+    this.minBackoff.set(minBackoff);
+  }
+
+  public long getMaxBackoff() {
+    return this.maxBackoff.get();
+  }
+
+  public void setMaxBackoff(long maxBackoff) {
+    this.maxBackoff.set(maxBackoff);
+  }
+
+  public long getTotalRequests() {
+    return this.totalRequests.get();
+  }
+
+  public void incrementTotalRequests() {
+    this.totalRequests.incrementAndGet();
+  }
+
+  public void setTotalRequests(long totalRequests) {
+    this.totalRequests.set(totalRequests);
+  }
+
+  public long getTotalBackoff() {
+    return this.totalBackoff.get();
+  }
+
+  public void setTotalBackoff(long totalBackoff) {
+    this.totalBackoff.set(totalBackoff);
+  }
+
+  public String getRetryCount() {
+    return this.retryCount;
+  }
+
+  public long getNumberOfIOPSThrottledRequests() {
+    return this.numberOfIOPSThrottledRequests.get();
+  }
+
+  public void setNumberOfIOPSThrottledRequests(long numberOfIOPSThrottledRequests) {
+    this.numberOfIOPSThrottledRequests.set(numberOfIOPSThrottledRequests);
+  }
+
+  public void incrementNumberOfIOPSThrottledRequests() {
+    this.numberOfIOPSThrottledRequests.getAndIncrement();
+  }
+
+  public long getNumberOfBandwidthThrottledRequests() {
+    return this.numberOfBandwidthThrottledRequests.get();
+  }
+
+  public void setNumberOfBandwidthThrottledRequests(long numberOfBandwidthThrottledRequests) {
+    this.numberOfBandwidthThrottledRequests.set(numberOfBandwidthThrottledRequests);
+  }
+
+  public void incrementNumberOfBandwidthThrottledRequests() {
+    this.numberOfBandwidthThrottledRequests.getAndIncrement();
+  }
+
+  public long getNumberOfOtherThrottledRequests() {
+    return this.numberOfOtherThrottledRequests.get();
+  }
+
+  public void setNumberOfOtherThrottledRequests(long numberOfOtherThrottledRequests) {
+    this.numberOfOtherThrottledRequests.set(numberOfOtherThrottledRequests);
+  }
+
+  public void incrementNumberOfOtherThrottledRequests() {
+    this.numberOfOtherThrottledRequests.getAndIncrement();
+  }
+
+  public long getMaxRetryCount() {
+    return this.maxRetryCount.get();
+  }
+
+  public void setMaxRetryCount(long maxRetryCount) {
+    this.maxRetryCount.set(maxRetryCount);
+  }
+
+  public void incrementMaxRetryCount() {
+    this.maxRetryCount.getAndIncrement();
+  }
+
+  public long getTotalNumberOfRequests() {
+    return this.totalNumberOfRequests.get();
+  }
+
+  public void setTotalNumberOfRequests(long totalNumberOfRequests) {
+    this.totalNumberOfRequests.set(totalNumberOfRequests);
+  }
+
+  public void incrementTotalNumberOfRequests() {
+    this.totalNumberOfRequests.getAndIncrement();
+  }
+
+  public Map<String, AbfsBackoffMetrics> getMetricsMap() {
+    return metricsMap;
+  }
+
+  public long getNumberOfRequestsSucceededWithoutRetrying() {
+    return this.numberOfRequestsSucceededWithoutRetrying.get();
+  }
+
+  public void setNumberOfRequestsSucceededWithoutRetrying(long numberOfRequestsSucceededWithoutRetrying) {
+    this.numberOfRequestsSucceededWithoutRetrying.set(numberOfRequestsSucceededWithoutRetrying);
+  }
+
+  public void incrementNumberOfRequestsSucceededWithoutRetrying() {
+    this.numberOfRequestsSucceededWithoutRetrying.getAndIncrement();
+  }
+
+  public long getNumberOfRequestsFailed() {
+    return this.numberOfRequestsFailed.get();
+  }
+
+  public void setNumberOfRequestsFailed(long numberOfRequestsFailed) {
+    this.numberOfRequestsFailed.set(numberOfRequestsFailed);
+  }
+
+  public void incrementNumberOfRequestsFailed() {
+    this.numberOfRequestsFailed.getAndIncrement();
+  }
+
+  public long getNumberOfNetworkFailedRequests() {
+    return this.numberOfNetworkFailedRequests.get();
+  }
+
+  public void setNumberOfNetworkFailedRequests(long numberOfNetworkFailedRequests) {
+    this.numberOfNetworkFailedRequests.set(numberOfNetworkFailedRequests);
+  }
+
+  public void incrementNumberOfNetworkFailedRequests() {
+    this.numberOfNetworkFailedRequests.getAndIncrement();
+  }
+
+  /*
+          Acronyms :-
+          1.RCTSI :- Request count that succeeded in x retries
+          2.MMA :- Min Max Average (This refers to the backoff or sleep time between 2 requests)
+          3.s :- seconds
+          4.BWT :- Number of Bandwidth throttled requests
+          5.IT :- Number of IOPS throttled requests
+          6.OT :- Number of Other throttled requests
+          7.NFR :- Number of requests which failed due to network errors
+          8.%RT :- Percentage of requests that are throttled
+          9.TRNR :- Total number of requests which succeeded without retrying
+          10.TRF :- Total number of requests which failed
+          11.TR :- Total number of requests which were made
+          12.MRC :- Max retry count across all requests
+           */
+  @Override
+  public String toString() {
+    StringBuilder metricString = new StringBuilder();
+    long totalRequestsThrottled = getNumberOfBandwidthThrottledRequests()
+        + getNumberOfIOPSThrottledRequests()
+        + getNumberOfOtherThrottledRequests();
+    double percentageOfRequestsThrottled =
+        ((double) totalRequestsThrottled / getTotalNumberOfRequests()) * HUNDRED;
+    for (Map.Entry<String, AbfsBackoffMetrics> entry : metricsMap.entrySet()) {

Review Comment:
   this is going to blow up if you used the string constructor.
   
   * consider adding a unit test calling .toString() immediately after creating the object.
   * if you use IOStatistics, as I've proposed, you can just use ioStatisticsToString to print these things



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -1731,6 +1792,128 @@ protected AccessTokenProvider getTokenProvider() {
     return tokenProvider;
   }
 
+  /**
+   * Retrieves a TracingContext object configured for metric tracking.
+   * This method creates a TracingContext object with the validated client correlation ID,
+   * the host name of the local machine (or "UnknownHost" if unable to determine),
+   * the file system operation type set to GET_ATTR, and additional configuration parameters
+   * for metric tracking.
+   * The TracingContext is intended for use in tracking metrics related to Azure Blob FileSystem (ABFS) operations.
+   *
+   * @return A TracingContext object configured for metric tracking.
+   */
+  private TracingContext getMetricTracingContext() {
+    String hostName;
+    try {
+      hostName = InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      hostName = "UnknownHost";
+    }
+    return new TracingContext(TracingContext.validateClientCorrelationID(
+        abfsConfiguration.getClientCorrelationId()),
+        hostName, FSOperationType.GET_ATTR, true,
+        abfsConfiguration.getTracingHeaderFormat(),
+        null, abfsCounters.toString());
+  }
+
+  /**
+   * Synchronized method to suspend or resume timer.
+   * @param timerFunctionality resume or suspend.
+   * @param timerTask The timertask object.
+   * @return true or false.
+   */
+  boolean timerOrchestrator(TimerFunctionality timerFunctionality, TimerTask timerTask) {
+    switch (timerFunctionality) {
+      case RESUME:
+        if (isMetricCollectionStopped.get()) {
+          synchronized (this) {
+            if (isMetricCollectionStopped.get()) {
+              resumeTimer();
+            }
+          }
+        }
+        break;
+      case SUSPEND:
+        long now = System.currentTimeMillis();
+        long lastExecutionTime = abfsCounters.getLastExecutionTime().get();
+        if (isMetricCollectionEnabled && (now - lastExecutionTime >= metricAnalysisPeriod)) {
+          synchronized (this) {
+            if (!isMetricCollectionStopped.get()) {
+              timerTask.cancel();
+              timer.purge();
+              isMetricCollectionStopped.set(true);
+              return true;
+            }
+          }
+        }
+        break;
+      default:
+        break;
+    }
+    return false;
+  }
+
+  private void resumeTimer() {
+    isMetricCollectionStopped.set(false);
+    timer.schedule(new TimerTaskImpl(),
+        metricIdlePeriod,
+        metricIdlePeriod);
+  }
+
+  /**
+   * Initiates a metric call to the Azure Blob FileSystem (ABFS) for retrieving file system properties.
+   * This method performs a HEAD request to the specified metric URL, using default headers and query parameters.
+   *
+   * @param tracingContext The tracing context to be used for capturing tracing information.
+   * @throws IOException throws IOException.
+   */
+  public void getMetricCall(TracingContext tracingContext) throws IOException {
+    this.isSendMetricCall = true;

Review Comment:
   no need for `this.`



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -25,6 +25,8 @@
 import java.net.UnknownHostException;
 import java.util.List;
 
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;

Review Comment:
   I'm just going to give up commenting on import ordering from here. You know what I'm going to say. Please: try and keep the imports under control. Your IDE can help you do this.



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";
+
+  /**
+   * Integration test for reading footer metrics with both Parquet and non-Parquet reads.
+   */
+  @Test
+  public void testReadFooterMetricsWithParquetAndNonParquet() throws Exception {
+    testReadWriteAndSeek(8 * ONE_MB, DEFAULT_READ_BUFFER_SIZE, ONE_KB, 4 * ONE_KB);
+  }
+
+  /**
+   * Configures the AzureBlobFileSystem with the given buffer size.
+   *
+   * @param bufferSize Buffer size to set for write and read operations.
+   * @return AbfsConfiguration used for configuration.
+   */
+  private Configuration getConfiguration(int bufferSize) {
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_FOOTER_METRIC_FORMAT));
+    configuration.setInt(AZURE_READ_BUFFER_SIZE, bufferSize);
+    configuration.setInt(AZURE_WRITE_BUFFER_SIZE, bufferSize);
+    return configuration;
+  }
+
+  /**
+   * Writes data to the specified file path in the AzureBlobFileSystem.
+   *
+   * @param fs      AzureBlobFileSystem instance.
+   * @param testPath Path to the file.
+   * @param data    Data to write to the file.
+   */
+  private void writeDataToFile(AzureBlobFileSystem fs, Path testPath, byte[] data) throws IOException {
+    FSDataOutputStream stream = fs.create(testPath);
+    try {
+      stream.write(data);
+    } finally {
+      stream.close();
+    }
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG, IOSTATISTICS_LOGGING_LEVEL_INFO, stream);
+  }
+
+  /**
+   * Asserts that the actual metrics obtained from the AzureBlobFileSystem match the expected metrics string.
+   *
+   * @param fs               AzureBlobFileSystem instance.
+   * @param expectedMetrics  Expected metrics string.
+   */
+  private void assertMetricsEquality(AzureBlobFileSystem fs, String expectedMetrics) {
+    AbfsReadFooterMetrics actualMetrics = fs.getAbfsClient().getAbfsCounters().getAbfsReadFooterMetrics();
+    assertNotNull("AbfsReadFooterMetrics is null", actualMetrics);
+    assertEquals("The computed metrics differs from the actual metrics", expectedMetrics, actualMetrics.toString());
+  }
+
+  /**
+   * Test for reading footer metrics with a non-Parquet file.
+   */
+  @Test
+  public void testReadFooterMetrics() throws Exception {
+    // Initialize AzureBlobFileSystem and set buffer size for configuration.
+    int bufferSize = MIN_BUFFER_SIZE;
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
+    AbfsConfiguration abfsConfiguration = fs.getAbfsStore().getAbfsConfiguration();
+
+    // Generate random data to write to the test file.
+    final byte[] b = new byte[2 * bufferSize];
+    new Random().nextBytes(b);
+
+    // Set up the test file path.
+    Path testPath = path(TEST_PATH);
+
+    // Write random data to the test file.
+    writeDataToFile(fs, testPath, b);
+
+    // Initialize a buffer for reading data.
+    final byte[] readBuffer = new byte[2 * bufferSize];
+    int result;
+
+    // Initialize statistics source for logging.
+    IOStatisticsSource statisticsSource = null;
+
+    try (FSDataInputStream inputStream = fs.open(testPath)) {
+      // Register a listener for tracing header validation.
+      statisticsSource = inputStream;
+      ((AbfsInputStream) inputStream.getWrappedStream()).registerListener(
+              new TracingHeaderValidator(abfsConfiguration.getClientCorrelationId(),
+                      fs.getFileSystemId(), FSOperationType.READ, true, 0,
+                      ((AbfsInputStream) inputStream.getWrappedStream())
+                              .getStreamID()));
+
+      // Perform the first read operation with seek.
+      inputStream.seek(bufferSize);
+      result = inputStream.read(readBuffer, bufferSize, bufferSize);
+      assertNotEquals(-1, result);
+
+      // To test tracingHeader for case with bypassReadAhead == true
+      inputStream.seek(0);
+      byte[] temp = new byte[5];
+      int t = inputStream.read(temp, 0, 1);
+
+      // Seek back to the beginning and perform another read operation.
+      inputStream.seek(0);
+      result = inputStream.read(readBuffer, 0, bufferSize);
+    }
+
+    // Log IO statistics at the INFO level.
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG,
+            IOSTATISTICS_LOGGING_LEVEL_INFO, statisticsSource);
+
+    // Ensure data is read successfully and matches the written data.
+    assertNotEquals("data read in final read()", -1, result);
+    assertArrayEquals(readBuffer, b);
+
+    // Get non-Parquet metrics and assert metrics equality.
+    AbfsReadFooterMetrics nonParquetMetrics = getNonParquetMetrics();
+    String metrics = nonParquetMetrics.getReadFooterMetrics(nonParquetMetrics);
+    assertMetricsEquality(fs, metrics);
+
+    // Close the AzureBlobFileSystem.
+    fs.close();
+  }
+
+  /**
+   * Generates and returns an instance of AbfsReadFooterMetrics for non-Parquet files.
+   */
+  private AbfsReadFooterMetrics getNonParquetMetrics() {
+    AbfsReadFooterMetrics nonParquetMetrics = new AbfsReadFooterMetrics();
+    nonParquetMetrics.setIsParquetFile(false);
+    nonParquetMetrics.setSizeReadByFirstRead("16384.000_16384.000");
+    nonParquetMetrics.setOffsetDiffBetweenFirstAndSecondRead("1.000_16384.000");
+    nonParquetMetrics.setAvgFileLength(Double.parseDouble("32768.000"));
+    nonParquetMetrics.setAvgReadLenRequested(Double.parseDouble("16384.000"));
+    return nonParquetMetrics;
+  }
+
+  /**
+   * Generates and returns an instance of AbfsReadFooterMetrics for parquet files.
+   */
+  private AbfsReadFooterMetrics getParquetMetrics() {
+    AbfsReadFooterMetrics parquetMetrics = new AbfsReadFooterMetrics();
+    parquetMetrics.setIsParquetFile(true);
+    parquetMetrics.setSizeReadByFirstRead("1024.000");
+    parquetMetrics.setOffsetDiffBetweenFirstAndSecondRead("4096.000");
+    parquetMetrics.setAvgFileLength(Double.parseDouble("8388608.000"));
+    parquetMetrics.setAvgReadLenRequested(0.000);
+    return parquetMetrics;
+  }
+
+  /**
+   * Test for reading, writing, and seeking with footer metrics.
+   *
+   * This method performs the integration test for reading, writing, and seeking operations
+   * with footer metrics. It creates an AzureBlobFileSystem, configures it, writes random data
+   * to a test file, performs read and seek operations, and checks the footer metrics for both
+   * Parquet and non-Parquet scenarios.
+   *
+   * @param fileSize Size of the test file.
+   * @param bufferSize Size of the buffer used for read and write operations.
+   * @param seek1 The position to seek to in the test file.
+   * @param seek2 Additional position to seek to in the test file (if not 0).
+   */
+  private void testReadWriteAndSeek(int fileSize, int bufferSize, Integer seek1, Integer seek2) throws Exception {
+    // Create an AzureBlobFileSystem instance.
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
+    AbfsConfiguration abfsConfiguration = fs.getAbfsStore().getAbfsConfiguration();
+
+    // Generate random data to write to the test file.
+    final byte[] b = new byte[fileSize];
+    new Random().nextBytes(b);
+
+    // Define the path for the test file.
+    Path testPath = path("/testfile");
+
+    // Write the random data to the test file.
+    writeDataToFile(fs, testPath, b);
+
+    // Initialize a buffer for reading.
+    final byte[] readBuffer = new byte[fileSize];
+
+    // Initialize a source for IO statistics.
+    IOStatisticsSource statisticsSource = null;
+
+    // Open an input stream for the test file.
+    FSDataInputStream inputStream = fs.open(testPath);
+    statisticsSource = inputStream;
+
+    // Register a listener for tracing headers.
+    ((AbfsInputStream) inputStream.getWrappedStream()).registerListener(
+            new TracingHeaderValidator(abfsConfiguration.getClientCorrelationId(),
+                    fs.getFileSystemId(), FSOperationType.READ, true, 0,
+                    ((AbfsInputStream) inputStream.getWrappedStream())
+                            .getStreamID()));
+
+    // Seek to the specified position in the test file and read data.
+    inputStream.seek(fileSize - seek1);
+    inputStream.read(readBuffer, 0, seek1);
+
+    // If seek2 is non-zero, perform an additional seek and read.
+    if (seek2 != 0) {
+      inputStream.seek(fileSize - seek1 - seek2);
+      inputStream.read(readBuffer, 0, seek2);
+    }
+
+    // Close the input stream.
+    inputStream.close();

Review Comment:
   close in a finally() clause



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -68,17 +76,20 @@ public class AbfsRestOperation {
   private final String sasToken;
 
   private static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);

Review Comment:
   so the wrong log was always used? not good. 
   Presumably, you don't want to move off it because that could break logging settings -correct?
   
   I would rename the existing log LOG_ABFS_CLIENT and at java doc to say why is being retained but it should not be used in any more places



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;

Review Comment:
   are these timings? if so should include time unit in name or at least use millis and javadoc it



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap
+      = new ConcurrentHashMap<>();
+
+  public AbfsBackoffMetrics() {
+    initializeMap();
+    this.numberOfIOPSThrottledRequests = new AtomicLong();
+    this.numberOfBandwidthThrottledRequests = new AtomicLong();
+    this.numberOfOtherThrottledRequests = new AtomicLong();
+    this.totalNumberOfRequests = new AtomicLong();
+    this.maxRetryCount = new AtomicLong();
+    this.numberOfRequestsSucceededWithoutRetrying = new AtomicLong();
+    this.numberOfRequestsFailed = new AtomicLong();
+    this.numberOfNetworkFailedRequests = new AtomicLong();
+  }
+
+  public AbfsBackoffMetrics(String retryCount) {
+    this.retryCount = retryCount;
+    this.numberOfRequestsSucceeded = new AtomicLong();
+    this.minBackoff = new AtomicLong(Long.MAX_VALUE);
+    this.maxBackoff = new AtomicLong();
+    this.totalRequests = new AtomicLong();
+    this.totalBackoff = new AtomicLong();
+  }
+
+  private void initializeMap() {
+    ArrayList<String> retryCountList = new ArrayList<String>(
+        Arrays.asList("1", "2", "3", "4", "5_15", "15_25", "25AndAbove"));
+    for (String s : retryCountList) {
+      metricsMap.put(s, new AbfsBackoffMetrics(s));
+    }
+  }
+
+  public long getNumberOfRequestsSucceeded() {
+    return this.numberOfRequestsSucceeded.get();
+  }
+
+  public void setNumberOfRequestsSucceeded(long numberOfRequestsSucceeded) {
+    this.numberOfRequestsSucceeded.set(numberOfRequestsSucceeded);
+  }
+
+  public void incrementNumberOfRequestsSucceeded() {
+    this.numberOfRequestsSucceeded.getAndIncrement();
+  }
+
+  public long getMinBackoff() {
+    return this.minBackoff.get();
+  }
+
+  public void setMinBackoff(long minBackoff) {
+    this.minBackoff.set(minBackoff);
+  }
+
+  public long getMaxBackoff() {
+    return this.maxBackoff.get();
+  }
+
+  public void setMaxBackoff(long maxBackoff) {
+    this.maxBackoff.set(maxBackoff);
+  }
+
+  public long getTotalRequests() {
+    return this.totalRequests.get();
+  }
+
+  public void incrementTotalRequests() {
+    this.totalRequests.incrementAndGet();
+  }
+
+  public void setTotalRequests(long totalRequests) {
+    this.totalRequests.set(totalRequests);
+  }
+
+  public long getTotalBackoff() {
+    return this.totalBackoff.get();
+  }
+
+  public void setTotalBackoff(long totalBackoff) {
+    this.totalBackoff.set(totalBackoff);
+  }
+
+  public String getRetryCount() {
+    return this.retryCount;
+  }
+
+  public long getNumberOfIOPSThrottledRequests() {
+    return this.numberOfIOPSThrottledRequests.get();
+  }
+
+  public void setNumberOfIOPSThrottledRequests(long numberOfIOPSThrottledRequests) {
+    this.numberOfIOPSThrottledRequests.set(numberOfIOPSThrottledRequests);
+  }
+
+  public void incrementNumberOfIOPSThrottledRequests() {
+    this.numberOfIOPSThrottledRequests.getAndIncrement();
+  }
+
+  public long getNumberOfBandwidthThrottledRequests() {
+    return this.numberOfBandwidthThrottledRequests.get();
+  }
+
+  public void setNumberOfBandwidthThrottledRequests(long numberOfBandwidthThrottledRequests) {
+    this.numberOfBandwidthThrottledRequests.set(numberOfBandwidthThrottledRequests);
+  }
+
+  public void incrementNumberOfBandwidthThrottledRequests() {
+    this.numberOfBandwidthThrottledRequests.getAndIncrement();
+  }
+
+  public long getNumberOfOtherThrottledRequests() {
+    return this.numberOfOtherThrottledRequests.get();
+  }
+
+  public void setNumberOfOtherThrottledRequests(long numberOfOtherThrottledRequests) {
+    this.numberOfOtherThrottledRequests.set(numberOfOtherThrottledRequests);
+  }
+
+  public void incrementNumberOfOtherThrottledRequests() {
+    this.numberOfOtherThrottledRequests.getAndIncrement();
+  }
+
+  public long getMaxRetryCount() {
+    return this.maxRetryCount.get();
+  }
+
+  public void setMaxRetryCount(long maxRetryCount) {
+    this.maxRetryCount.set(maxRetryCount);
+  }
+
+  public void incrementMaxRetryCount() {
+    this.maxRetryCount.getAndIncrement();
+  }
+
+  public long getTotalNumberOfRequests() {
+    return this.totalNumberOfRequests.get();
+  }
+
+  public void setTotalNumberOfRequests(long totalNumberOfRequests) {
+    this.totalNumberOfRequests.set(totalNumberOfRequests);
+  }
+
+  public void incrementTotalNumberOfRequests() {
+    this.totalNumberOfRequests.getAndIncrement();
+  }
+
+  public Map<String, AbfsBackoffMetrics> getMetricsMap() {
+    return metricsMap;
+  }
+
+  public long getNumberOfRequestsSucceededWithoutRetrying() {
+    return this.numberOfRequestsSucceededWithoutRetrying.get();
+  }
+
+  public void setNumberOfRequestsSucceededWithoutRetrying(long numberOfRequestsSucceededWithoutRetrying) {
+    this.numberOfRequestsSucceededWithoutRetrying.set(numberOfRequestsSucceededWithoutRetrying);
+  }
+
+  public void incrementNumberOfRequestsSucceededWithoutRetrying() {
+    this.numberOfRequestsSucceededWithoutRetrying.getAndIncrement();
+  }
+
+  public long getNumberOfRequestsFailed() {
+    return this.numberOfRequestsFailed.get();
+  }
+
+  public void setNumberOfRequestsFailed(long numberOfRequestsFailed) {
+    this.numberOfRequestsFailed.set(numberOfRequestsFailed);
+  }
+
+  public void incrementNumberOfRequestsFailed() {
+    this.numberOfRequestsFailed.getAndIncrement();
+  }
+
+  public long getNumberOfNetworkFailedRequests() {
+    return this.numberOfNetworkFailedRequests.get();
+  }
+
+  public void setNumberOfNetworkFailedRequests(long numberOfNetworkFailedRequests) {
+    this.numberOfNetworkFailedRequests.set(numberOfNetworkFailedRequests);
+  }
+
+  public void incrementNumberOfNetworkFailedRequests() {
+    this.numberOfNetworkFailedRequests.getAndIncrement();
+  }
+
+  /*
+          Acronyms :-
+          1.RCTSI :- Request count that succeeded in x retries
+          2.MMA :- Min Max Average (This refers to the backoff or sleep time between 2 requests)
+          3.s :- seconds
+          4.BWT :- Number of Bandwidth throttled requests
+          5.IT :- Number of IOPS throttled requests
+          6.OT :- Number of Other throttled requests
+          7.NFR :- Number of requests which failed due to network errors
+          8.%RT :- Percentage of requests that are throttled
+          9.TRNR :- Total number of requests which succeeded without retrying
+          10.TRF :- Total number of requests which failed
+          11.TR :- Total number of requests which were made
+          12.MRC :- Max retry count across all requests
+           */
+  @Override
+  public String toString() {

Review Comment:
   What is the goal here? something for logging or to submit as a header in a message.
   
   And do you expect the log entries to be structured enough for parsing later?
   
   If so, I would propose having a toLogString() message and highlight that it must not be changed



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsCountersImpl.java:
##########
@@ -63,6 +66,12 @@ public class AbfsCountersImpl implements AbfsCounters {
 
   private final IOStatisticsStore ioStatisticsStore;
 
+  private AbfsBackoffMetrics abfsBackoffMetrics = null;

Review Comment:
   don't bother with the = null here and below



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/services/TestAbfsRestOperation.java:
##########
@@ -0,0 +1,81 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.AbfsConfiguration;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.HTTP_METHOD_DELETE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.services.AbfsRestOperationType.DeletePath;
+import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem;
+import org.apache.hadoop.fs.azurebfs.AbstractAbfsIntegrationTest;
+import java.util.ArrayList;
+import java.util.Arrays;
+import org.junit.Assert;
+import java.net.HttpURLConnection;
+
+public class TestAbfsRestOperation extends
+    AbstractAbfsIntegrationTest {
+
+  public TestAbfsRestOperation() throws Exception {
+  }
+
+  /**
+   * Test for backoff retry metrics.
+   *
+   * This method tests backoff retry metrics by creating an AzureBlobFileSystem, initializing an
+   * AbfsClient, and performing mock operations on an AbfsRestOperation. The method then updates
+   * backoff metrics using the AbfsRestOperation.
+   *
+   */
+  @Test
+  public void testBackoffRetryMetrics() throws Exception {
+    // Create an AzureBlobFileSystem instance.
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_BACKOFF_METRIC_FORMAT));
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);

Review Comment:
   try/finally. always.



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java:
##########
@@ -60,6 +61,9 @@ public enum AzureServiceErrorCode {
   private final String errorCode;
   private final int httpStatusCode;
   private final String errorMessage;
+
+  private static final Logger LOG1 = LoggerFactory.getLogger(AzureServiceErrorCode.class);

Review Comment:
   call it LOG



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap

Review Comment:
   this is mixing up a metrics class "stats on use" with the map of filenames -> instances. it explains why there are two constructors and one only does partial init and whose toString() will NPE -but it doesn't justify this design.
   
   Proposed: split out the metrics (which should use IOStatisticsStore for its structure, and implement IOStatisticsSource serving this) from the map managing the metrics.
   
   that must be at most one per client instance and needs a good cleanup story so it scales well.



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -315,7 +373,35 @@ private boolean executeHttpOperation(final int retryCount,
       }
 
       httpOperation.processResponse(buffer, bufferOffset, bufferLength);
-      incrementCounter(AbfsStatistic.GET_RESPONSES, 1);
+      if (!isThrottledRequest && httpOperation.getStatusCode()
+          >= HttpURLConnection.HTTP_INTERNAL_ERROR) {
+        isThrottledRequest = true;
+        AzureServiceErrorCode serviceErrorCode =
+            AzureServiceErrorCode.getAzureServiceCode(
+                httpOperation.getStatusCode(),
+                httpOperation.getStorageErrorCode(),
+                httpOperation.getStorageErrorMessage());
+        LOG1.trace("Service code is " + serviceErrorCode + " status code is "

Review Comment:
   use {} for insertion so that if trace is disabled no strings are built up



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -351,7 +437,13 @@ private boolean executeHttpOperation(final int retryCount,
       retryPolicy = client.getRetryPolicy(failureReason);
       LOG.warn("Unknown host name: {}. Retrying to resolve the host name...",
           hostname);
+      if (abfsBackoffMetrics != null) {

Review Comment:
   factor out into own method and use in all invocations



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {

Review Comment:
   can you add javadocs for each of these and the class.



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -206,11 +228,12 @@ String getSasToken() {
    */
   public void execute(TracingContext tracingContext)
       throws AzureBlobFileSystemException {
-
     // Since this might be a sub-sequential or parallel rest operation
     // triggered by a single file system call, using a new tracing context.
     lastUsedTracingContext = createNewTracingContext(tracingContext);
     try {
+      abfsCounters.getLastExecutionTime().set(now());
+      client.timerOrchestrator(TimerFunctionality.RESUME, null);

Review Comment:
   This is getting really convoluted now. I believe the callbacks to be made of AbfsClient should be passed down as an interface so you can stay in control of what is being asked of it. It should also help testing



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -68,17 +76,20 @@ public class AbfsRestOperation {
   private final String sasToken;
 
   private static final Logger LOG = LoggerFactory.getLogger(AbfsClient.class);
-
+  private static final Logger LOG1 = LoggerFactory.getLogger(AbfsRestOperation.class);

Review Comment:
   ...and this can be called `LOG`



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,549 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.Map;
+import java.util.StringJoiner;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.List;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+
+public class AbfsReadFooterMetrics {
+  private final AtomicBoolean isParquetFile;
+  private final AtomicBoolean isParquetEvaluated;
+  private final AtomicBoolean isLenUpdated;
+  private String sizeReadByFirstRead;
+  private String offsetDiffBetweenFirstAndSecondRead;
+  private final AtomicLong fileLength;
+  private double avgFileLength;
+  private double avgReadLenRequested;
+  private final AtomicBoolean collectMetrics;
+  private final AtomicBoolean collectMetricsForNextRead;
+  private final AtomicBoolean collectLenMetrics;
+  private final AtomicLong dataLenRequested;
+  private final AtomicLong offsetOfFirstRead;
+  private final AtomicInteger readCount;
+  private final ConcurrentSkipListMap<String, AbfsReadFooterMetrics> metricsMap;
+  private static final String FOOTER_LENGTH = "20";
+
+  public AbfsReadFooterMetrics() {
+    this.isParquetFile = new AtomicBoolean(false);
+    this.isParquetEvaluated = new AtomicBoolean(false);
+    this.isLenUpdated = new AtomicBoolean(false);
+    this.fileLength = new AtomicLong();
+    this.readCount = new AtomicInteger(0);
+    this.offsetOfFirstRead = new AtomicLong();
+    this.collectMetrics = new AtomicBoolean(false);
+    this.collectMetricsForNextRead = new AtomicBoolean(false);
+    this.collectLenMetrics = new AtomicBoolean(false);
+    this.dataLenRequested = new AtomicLong(0);
+    this.metricsMap = new ConcurrentSkipListMap<>();
+  }
+
+  public Map<String, AbfsReadFooterMetrics> getMetricsMap() {
+    return metricsMap;
+  }
+
+  private boolean getIsParquetFile() {
+    return isParquetFile.get();
+  }
+
+  public void setIsParquetFile(boolean isParquetFile) {
+    this.isParquetFile.set(isParquetFile);
+  }
+
+  private String getSizeReadByFirstRead() {
+    return sizeReadByFirstRead;
+  }
+
+  public void setSizeReadByFirstRead(final String sizeReadByFirstRead) {
+    this.sizeReadByFirstRead = sizeReadByFirstRead;
+  }
+
+  private String getOffsetDiffBetweenFirstAndSecondRead() {
+    return offsetDiffBetweenFirstAndSecondRead;
+  }
+
+  public void setOffsetDiffBetweenFirstAndSecondRead(final String offsetDiffBetweenFirstAndSecondRead) {
+    this.offsetDiffBetweenFirstAndSecondRead
+        = offsetDiffBetweenFirstAndSecondRead;
+  }
+
+  private long getFileLength() {
+    return fileLength.get();
+  }
+
+  private void setFileLength(long fileLength) {
+    this.fileLength.set(fileLength);
+  }
+
+  private double getAvgFileLength() {
+    return avgFileLength;
+  }
+
+  public void setAvgFileLength(final double avgFileLength) {
+    this.avgFileLength = avgFileLength;
+  }
+
+  private double getAvgReadLenRequested() {
+    return avgReadLenRequested;
+  }
+
+  public void setAvgReadLenRequested(final double avgReadLenRequested) {
+    this.avgReadLenRequested = avgReadLenRequested;
+  }
+
+  private boolean getCollectMetricsForNextRead() {
+    return collectMetricsForNextRead.get();
+  }
+
+  private void setCollectMetricsForNextRead(boolean collectMetricsForNextRead) {
+    this.collectMetricsForNextRead.set(collectMetricsForNextRead);
+  }
+
+  private long getOffsetOfFirstRead() {
+    return offsetOfFirstRead.get();
+  }
+
+  private void setOffsetOfFirstRead(long offsetOfFirstRead) {
+    this.offsetOfFirstRead.set(offsetOfFirstRead);
+  }
+
+  private int getReadCount() {
+    return readCount.get();
+  }
+
+  private void setReadCount(int readCount) {
+    this.readCount.set(readCount);
+  }
+
+  private int incrementReadCount() {
+    this.readCount.incrementAndGet();
+    return getReadCount();
+  }
+
+  private boolean getCollectLenMetrics() {
+    return collectLenMetrics.get();
+  }
+
+  private void setCollectLenMetrics(boolean collectLenMetrics) {
+    this.collectLenMetrics.set(collectLenMetrics);
+
+  }
+
+  private long getDataLenRequested() {
+    return dataLenRequested.get();
+  }
+
+  private void setDataLenRequested(long dataLenRequested) {
+    this.dataLenRequested.set(dataLenRequested);
+  }
+
+  private void updateDataLenRequested(long dataLenRequested){
+    this.dataLenRequested.addAndGet(dataLenRequested);
+  }
+
+  private boolean getCollectMetrics() {
+    return collectMetrics.get();
+  }
+
+  private void setCollectMetrics(boolean collectMetrics) {
+    this.collectMetrics.set(collectMetrics);
+  }
+
+  private boolean getIsParquetEvaluated() {
+    return isParquetEvaluated.get();
+  }
+
+  private void setIsParquetEvaluated(boolean isParquetEvaluated) {
+    this.isParquetEvaluated.set(isParquetEvaluated);
+  }
+
+  private boolean getIsLenUpdated() {
+    return isLenUpdated.get();
+  }
+
+  private void setIsLenUpdated(boolean isLenUpdated) {
+    this.isLenUpdated.set(isLenUpdated);
+  }
+
+  /**
+   * Updates the metrics map with an entry for the specified file if it doesn't already exist.
+   *
+   * @param filePathIdentifier The unique identifier for the file.
+   */
+  public void updateMap(String filePathIdentifier) {
+    // If the file is not already in the metrics map, add it with a new AbfsReadFooterMetrics object.
+    metricsMap.computeIfAbsent(filePathIdentifier, key -> new AbfsReadFooterMetrics());
+  }
+
+  /**
+   * Checks and updates metrics for a specific file identified by filePathIdentifier.
+   * If the metrics do not exist for the file, they are initialized.
+   *
+   * @param filePathIdentifier The unique identifier for the file.
+   * @param len               The length of the read operation.
+   * @param contentLength     The total content length of the file.
+   * @param nextReadPos       The position of the next read operation.
+   */
+  public void checkMetricUpdate(final String filePathIdentifier, final int len, final long contentLength,
+      final long nextReadPos) {
+    AbfsReadFooterMetrics readFooterMetrics = metricsMap.computeIfAbsent(
+            filePathIdentifier, key -> new AbfsReadFooterMetrics());
+    if (readFooterMetrics.getReadCount() == 0
+        || (readFooterMetrics.getReadCount() >= 1
+        && readFooterMetrics.getCollectMetrics())) {
+      updateMetrics(filePathIdentifier, len, contentLength, nextReadPos);
+    }
+  }
+
+  /**
+   * Updates metrics for a specific file identified by filePathIdentifier.
+   *
+   * @param filePathIdentifier  The unique identifier for the file.
+   * @param len                The length of the read operation.
+   * @param contentLength      The total content length of the file.
+   * @param nextReadPos        The position of the next read operation.
+   */
+  private void updateMetrics(final String filePathIdentifier, final int len, final long contentLength,
+                             final long nextReadPos) {
+    AbfsReadFooterMetrics readFooterMetrics = metricsMap.get(filePathIdentifier);
+
+    // Create a new AbfsReadFooterMetrics object if it doesn't exist in the metricsMap.
+    if (readFooterMetrics == null) {
+      readFooterMetrics = new AbfsReadFooterMetrics();
+      metricsMap.put(filePathIdentifier, readFooterMetrics);
+    }
+
+    int readCount;
+    synchronized (this) {
+      readCount = readFooterMetrics.incrementReadCount();
+    }
+
+    if (readCount == 1) {
+      // Update metrics for the first read.
+      updateMetricsOnFirstRead(readFooterMetrics, nextReadPos, len, contentLength);
+    }
+
+    synchronized (this) {
+      if (readFooterMetrics.getCollectLenMetrics()) {
+        readFooterMetrics.updateDataLenRequested(len);
+      }
+    }
+
+    if (readCount == 2) {
+      // Update metrics for the second read.
+      updateMetricsOnSecondRead(readFooterMetrics, nextReadPos, len);
+    }
+  }
+
+  /**
+   * Updates metrics for the first read operation.
+   *
+   * @param readFooterMetrics The metrics object to update.
+   * @param nextReadPos       The position of the next read operation.
+   * @param len               The length of the read operation.
+   * @param contentLength     The total content length of the file.
+   */
+  private void updateMetricsOnFirstRead(AbfsReadFooterMetrics readFooterMetrics, long nextReadPos, int len, long contentLength) {
+    if (nextReadPos >= contentLength - (long) Integer.parseInt(FOOTER_LENGTH) * ONE_KB) {
+      readFooterMetrics.setCollectMetrics(true);
+      readFooterMetrics.setCollectMetricsForNextRead(true);
+      readFooterMetrics.setOffsetOfFirstRead(nextReadPos);
+      readFooterMetrics.setSizeReadByFirstRead(len + "_" + Math.abs(contentLength - nextReadPos));
+      readFooterMetrics.setFileLength(contentLength);
+    }
+  }
+
+  /**
+   * Updates metrics for the second read operation.
+   *
+   * @param readFooterMetrics The metrics object to update.
+   * @param nextReadPos       The position of the next read operation.
+   * @param len               The length of the read operation.
+   */
+  private void updateMetricsOnSecondRead(AbfsReadFooterMetrics readFooterMetrics, long nextReadPos, int len) {
+    if (readFooterMetrics.getCollectMetricsForNextRead()) {
+      long offsetDiff = Math.abs(nextReadPos - readFooterMetrics.getOffsetOfFirstRead());
+      readFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(len + "_" + offsetDiff);
+      readFooterMetrics.setCollectLenMetrics(true);
+    }
+  }
+
+
+  /**
+   * Check if the given file should be marked as a Parquet file.
+   *
+   * @param metrics The metrics to evaluate.
+   * @return True if the file meet the criteria for being marked as a Parquet file, false otherwise.
+   */
+  private boolean shouldMarkAsParquet(AbfsReadFooterMetrics metrics) {
+    return metrics.getCollectMetrics()
+            && metrics.getReadCount() >= 2
+            && !metrics.getIsParquetEvaluated()
+            && haveEqualValues(metrics.getSizeReadByFirstRead())
+            && haveEqualValues(metrics.getOffsetDiffBetweenFirstAndSecondRead());
+  }
+
+  /**
+   * Check if two values are equal, considering they are in the format "value1_value2".
+   *
+   * @param value The value to check.
+   * @return True if the two parts of the value are equal, false otherwise.
+   */
+  private boolean haveEqualValues(String value) {
+    String[] parts = value.split("_");
+    return parts.length == 2 && parts[0].equals(parts[1]);
+  }
+
+  /**
+   * Mark the given metrics as a Parquet file and update related values.
+   *
+   * @param metrics The metrics to mark as Parquet.
+   */
+  private void markAsParquet(AbfsReadFooterMetrics metrics) {
+    metrics.setIsParquetFile(true);
+    String[] parts = metrics.getSizeReadByFirstRead().split("_");
+    metrics.setSizeReadByFirstRead(parts[0]);
+    parts = metrics.getOffsetDiffBetweenFirstAndSecondRead().split("_");
+    metrics.setOffsetDiffBetweenFirstAndSecondRead(parts[0]);
+    metrics.setIsParquetEvaluated(true);
+  }
+
+  /**
+   * Check each metric in the provided map and mark them as Parquet files if they meet the criteria.
+   *
+   * @param metricsMap The map containing metrics to evaluate.
+   */
+  public void checkIsParquet(Map<String, AbfsReadFooterMetrics> metricsMap) {
+    for (Map.Entry<String, AbfsReadFooterMetrics> entry : metricsMap.entrySet()) {
+      AbfsReadFooterMetrics readFooterMetrics = entry.getValue();
+      if (shouldMarkAsParquet(readFooterMetrics)) {
+        markAsParquet(readFooterMetrics);
+        metricsMap.replace(entry.getKey(), readFooterMetrics);
+      }
+    }
+  }
+
+  /**
+   * Updates the average read length requested for metrics of all files in the metrics map.
+   * If the metrics indicate that the update is needed, it calculates the average read length and updates the metrics.
+   *
+   * @param metricsMap A map containing metrics for different files with unique identifiers.
+   */
+  private void updateLenRequested(Map<String, AbfsReadFooterMetrics> metricsMap) {
+    for (AbfsReadFooterMetrics readFooterMetrics : metricsMap.values()) {
+      if (shouldUpdateLenRequested(readFooterMetrics)) {
+        int readReqCount = readFooterMetrics.getReadCount() - 2;
+        readFooterMetrics.setAvgReadLenRequested(
+                (double) readFooterMetrics.getDataLenRequested() / readReqCount);
+        readFooterMetrics.setIsLenUpdated(true);
+      }
+    }
+  }
+
+  /**
+   * Checks whether the average read length requested should be updated for the given metrics.
+   *
+   * The method returns true if the following conditions are met:
+   * - Metrics collection is enabled.
+   * - The number of read counts is greater than 2.
+   * - The average read length has not been updated previously.
+   *
+   * @param readFooterMetrics The metrics object to evaluate.
+   * @return True if the average read length should be updated, false otherwise.
+   */
+  private boolean shouldUpdateLenRequested(AbfsReadFooterMetrics readFooterMetrics) {
+    return readFooterMetrics.getCollectMetrics()
+            && readFooterMetrics.getReadCount() > 2
+            && !readFooterMetrics.getIsLenUpdated();
+  }
+
+  /**
+   * Calculates the average metrics from a list of AbfsReadFooterMetrics and sets the values in the provided 'avgParquetReadFooterMetrics' object.
+   *
+   * @param isParquetList The list of AbfsReadFooterMetrics to compute the averages from.
+   * @param avgParquetReadFooterMetrics The target AbfsReadFooterMetrics object to store the computed average values.
+   *
+   * This method calculates various average metrics from the provided list and sets them in the 'avgParquetReadFooterMetrics' object.
+   * The metrics include:
+   * - Size read by the first read
+   * - Offset difference between the first and second read
+   * - Average file length
+   * - Average requested read length
+   */
+  private void getParquetReadFooterMetricsAverage(List<AbfsReadFooterMetrics> isParquetList,
+      AbfsReadFooterMetrics avgParquetReadFooterMetrics){
+    avgParquetReadFooterMetrics.setSizeReadByFirstRead(
+        String.format("%.3f", isParquetList.stream()
+            .map(AbfsReadFooterMetrics::getSizeReadByFirstRead).mapToDouble(
+                Double::parseDouble).average().orElse(0.0)));
+    avgParquetReadFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(
+        String.format("%.3f", isParquetList.stream()
+            .map(AbfsReadFooterMetrics::getOffsetDiffBetweenFirstAndSecondRead)
+            .mapToDouble(Double::parseDouble).average().orElse(0.0)));
+    avgParquetReadFooterMetrics.setAvgFileLength(isParquetList.stream()
+        .mapToDouble(AbfsReadFooterMetrics::getFileLength).average().orElse(0.0));
+    avgParquetReadFooterMetrics.setAvgReadLenRequested(isParquetList.stream().
+        map(AbfsReadFooterMetrics::getAvgReadLenRequested).
+        mapToDouble(Double::doubleValue).average().orElse(0.0));
+  }
+
+  /**
+   * Calculates the average metrics from a list of non-Parquet AbfsReadFooterMetrics instances.
+   *
+   * This method takes a list of AbfsReadFooterMetrics representing non-Parquet reads and calculates
+   * the average values for the size read by the first read and the offset difference between the first
+   * and second read. The averages are then set in the provided AbfsReadFooterMetrics instance.
+   *
+   * @param isNonParquetList A list of AbfsReadFooterMetrics instances representing non-Parquet reads.
+   * @param avgNonParquetReadFooterMetrics The AbfsReadFooterMetrics instance to store the calculated averages.
+   *                                      It is assumed that the size of the list is at least 1, and the first
+   *                                      element of the list is used to determine the size of arrays.
+   *                                      The instance is modified in-place with the calculated averages.
+   *
+   *
+   **/
+  private void getNonParquetReadFooterMetricsAverage(List<AbfsReadFooterMetrics> isNonParquetList,
+                                                     AbfsReadFooterMetrics avgNonParquetReadFooterMetrics) {
+    int size = isNonParquetList.get(0).getSizeReadByFirstRead().split("_").length;
+    double[] store = new double[2 * size];
+    // Calculating sum of individual values
+    isNonParquetList.forEach(abfsReadFooterMetrics -> {
+      String[] firstReadSize = abfsReadFooterMetrics.getSizeReadByFirstRead().split("_");
+      String[] offDiffFirstSecondRead = abfsReadFooterMetrics.getOffsetDiffBetweenFirstAndSecondRead().split("_");
+
+      for (int i = 0; i < firstReadSize.length; i++) {
+        store[i] += Long.parseLong(firstReadSize[i]);
+        store[i + size] += Long.parseLong(offDiffFirstSecondRead[i]);
+      }
+    });
+
+    // Calculating averages and creating formatted strings
+    StringJoiner firstReadSize = new StringJoiner("_");
+    StringJoiner offDiffFirstSecondRead = new StringJoiner("_");
+
+    for (int j = 0; j < size; j++) {
+      firstReadSize.add(String.format("%.3f", store[j] / isNonParquetList.size()));
+      offDiffFirstSecondRead.add(String.format("%.3f", store[j + size] / isNonParquetList.size()));
+    }
+
+    avgNonParquetReadFooterMetrics.setSizeReadByFirstRead(firstReadSize.toString());
+    avgNonParquetReadFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(offDiffFirstSecondRead.toString());
+    avgNonParquetReadFooterMetrics.setAvgFileLength(isNonParquetList.stream()
+            .mapToDouble(AbfsReadFooterMetrics::getFileLength).average().orElse(0.0));
+    avgNonParquetReadFooterMetrics.setAvgReadLenRequested(isNonParquetList.stream()
+            .mapToDouble(AbfsReadFooterMetrics::getAvgReadLenRequested).average().orElse(0.0));
+  }
+
+  /*
+  Acronyms:
+  1.FR :- First Read (In case of parquet we only maintain the size requested by application for
+  the first read, in case of non parquet we maintain a string separated by "_" delimiter where the first
+  substring represents the len requested for first read and the second substring represents the seek pointer difference from the
+  end of the file.)
+  2.SR :- Second Read (In case of parquet we only maintain the size requested by application for
+  the second read, in case of non parquet we maintain a string separated by "_" delimiter where the first
+  substring represents the len requested for second read and the second substring represents the seek pointer difference from the
+  offset of the first read.)
+  3.FL :- Total length of the file requested for read
+   */
+  public String getReadFooterMetrics(AbfsReadFooterMetrics avgReadFooterMetrics) {
+    String readFooterMetric = "";
+    if (avgReadFooterMetrics.getIsParquetFile()) {
+      readFooterMetric += "$Parquet:";
+    } else {
+      readFooterMetric += "$NonParquet:";
+    }
+    readFooterMetric += "$FR=" + avgReadFooterMetrics.getSizeReadByFirstRead()
+        + "$SR="
+        + avgReadFooterMetrics.getOffsetDiffBetweenFirstAndSecondRead()
+        + "$FL=" + String.format("%.3f",
+        avgReadFooterMetrics.getAvgFileLength())
+        + "$RL=" + String.format("%.3f",
+        avgReadFooterMetrics.getAvgReadLenRequested());
+    return readFooterMetric;
+  }
+
+/**
+ * Retrieves and aggregates read footer metrics for both Parquet and non-Parquet files from a list
+ * of AbfsReadFooterMetrics instances. The function calculates the average metrics separately for
+ * Parquet and non-Parquet files and returns a formatted string containing the aggregated metrics.
+ *
+ * @param readFooterMetricsList A list of AbfsReadFooterMetrics instances containing read footer metrics
+ *                              for both Parquet and non-Parquet files.
+ *
+ * @return A formatted string containing the aggregated read footer metrics for both Parquet and non-Parquet files.
+ *
+ **/
+private String getFooterMetrics(List<AbfsReadFooterMetrics> readFooterMetricsList) {
+  List<AbfsReadFooterMetrics> isParquetList = new ArrayList<>();
+  List<AbfsReadFooterMetrics> isNonParquetList = new ArrayList<>();
+  for (AbfsReadFooterMetrics abfsReadFooterMetrics : readFooterMetricsList) {
+    if (abfsReadFooterMetrics.getIsParquetFile()) {
+      isParquetList.add(abfsReadFooterMetrics);
+    } else {
+      if (abfsReadFooterMetrics.getReadCount() >= 2) {
+        isNonParquetList.add(abfsReadFooterMetrics);
+      }
+    }
+  }
+  AbfsReadFooterMetrics avgParquetReadFooterMetrics = new AbfsReadFooterMetrics();
+  AbfsReadFooterMetrics avgNonparquetReadFooterMetrics = new AbfsReadFooterMetrics();
+  String readFooterMetric = "";
+  if (!isParquetList.isEmpty()) {
+    avgParquetReadFooterMetrics.setIsParquetFile(true);
+    getParquetReadFooterMetricsAverage(isParquetList, avgParquetReadFooterMetrics);
+    readFooterMetric += getReadFooterMetrics(avgParquetReadFooterMetrics);
+  }
+  if (!isNonParquetList.isEmpty()) {
+    avgNonparquetReadFooterMetrics.setIsParquetFile(false);
+    getNonParquetReadFooterMetricsAverage(isNonParquetList, avgNonparquetReadFooterMetrics);
+    readFooterMetric += getReadFooterMetrics(avgNonparquetReadFooterMetrics);
+  }
+  return readFooterMetric;
+}
+
+
+  @Override
+  public String toString() {

Review Comment:
   i think this should be an explicit method, not toString() which is generally assumed to be flor logging and lower cost.



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java:
##########
@@ -700,6 +698,18 @@ public synchronized void close() throws IOException {
     if (isClosed) {
       return;
     }
+    if (abfsStore.getClient().isMetricCollectionEnabled()) {
+      TracingContext tracingMetricContext = new TracingContext(
+              clientCorrelationId,
+              fileSystemId, FSOperationType.GET_ATTR, true,

Review Comment:
   have a look at the `CommonAuditContext`; it includes things like jobid, kerberos principal and more...so letting you identify jobs and people/applications



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,549 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.Map;
+import java.util.StringJoiner;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.List;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+
+public class AbfsReadFooterMetrics {
+  private final AtomicBoolean isParquetFile;
+  private final AtomicBoolean isParquetEvaluated;
+  private final AtomicBoolean isLenUpdated;
+  private String sizeReadByFirstRead;
+  private String offsetDiffBetweenFirstAndSecondRead;
+  private final AtomicLong fileLength;
+  private double avgFileLength;
+  private double avgReadLenRequested;
+  private final AtomicBoolean collectMetrics;
+  private final AtomicBoolean collectMetricsForNextRead;
+  private final AtomicBoolean collectLenMetrics;
+  private final AtomicLong dataLenRequested;
+  private final AtomicLong offsetOfFirstRead;
+  private final AtomicInteger readCount;
+  private final ConcurrentSkipListMap<String, AbfsReadFooterMetrics> metricsMap;
+  private static final String FOOTER_LENGTH = "20";
+
+  public AbfsReadFooterMetrics() {
+    this.isParquetFile = new AtomicBoolean(false);
+    this.isParquetEvaluated = new AtomicBoolean(false);
+    this.isLenUpdated = new AtomicBoolean(false);
+    this.fileLength = new AtomicLong();
+    this.readCount = new AtomicInteger(0);
+    this.offsetOfFirstRead = new AtomicLong();
+    this.collectMetrics = new AtomicBoolean(false);
+    this.collectMetricsForNextRead = new AtomicBoolean(false);
+    this.collectLenMetrics = new AtomicBoolean(false);
+    this.dataLenRequested = new AtomicLong(0);
+    this.metricsMap = new ConcurrentSkipListMap<>();
+  }
+
+  public Map<String, AbfsReadFooterMetrics> getMetricsMap() {
+    return metricsMap;
+  }
+
+  private boolean getIsParquetFile() {
+    return isParquetFile.get();
+  }
+
+  public void setIsParquetFile(boolean isParquetFile) {
+    this.isParquetFile.set(isParquetFile);
+  }
+
+  private String getSizeReadByFirstRead() {
+    return sizeReadByFirstRead;
+  }
+
+  public void setSizeReadByFirstRead(final String sizeReadByFirstRead) {
+    this.sizeReadByFirstRead = sizeReadByFirstRead;
+  }
+
+  private String getOffsetDiffBetweenFirstAndSecondRead() {
+    return offsetDiffBetweenFirstAndSecondRead;
+  }
+
+  public void setOffsetDiffBetweenFirstAndSecondRead(final String offsetDiffBetweenFirstAndSecondRead) {
+    this.offsetDiffBetweenFirstAndSecondRead
+        = offsetDiffBetweenFirstAndSecondRead;
+  }
+
+  private long getFileLength() {
+    return fileLength.get();
+  }
+
+  private void setFileLength(long fileLength) {
+    this.fileLength.set(fileLength);
+  }
+
+  private double getAvgFileLength() {
+    return avgFileLength;
+  }
+
+  public void setAvgFileLength(final double avgFileLength) {
+    this.avgFileLength = avgFileLength;
+  }
+
+  private double getAvgReadLenRequested() {
+    return avgReadLenRequested;
+  }
+
+  public void setAvgReadLenRequested(final double avgReadLenRequested) {
+    this.avgReadLenRequested = avgReadLenRequested;
+  }
+
+  private boolean getCollectMetricsForNextRead() {
+    return collectMetricsForNextRead.get();
+  }
+
+  private void setCollectMetricsForNextRead(boolean collectMetricsForNextRead) {
+    this.collectMetricsForNextRead.set(collectMetricsForNextRead);
+  }
+
+  private long getOffsetOfFirstRead() {
+    return offsetOfFirstRead.get();
+  }
+
+  private void setOffsetOfFirstRead(long offsetOfFirstRead) {
+    this.offsetOfFirstRead.set(offsetOfFirstRead);
+  }
+
+  private int getReadCount() {
+    return readCount.get();
+  }
+
+  private void setReadCount(int readCount) {
+    this.readCount.set(readCount);
+  }
+
+  private int incrementReadCount() {
+    this.readCount.incrementAndGet();
+    return getReadCount();
+  }
+
+  private boolean getCollectLenMetrics() {
+    return collectLenMetrics.get();
+  }
+
+  private void setCollectLenMetrics(boolean collectLenMetrics) {
+    this.collectLenMetrics.set(collectLenMetrics);
+
+  }
+
+  private long getDataLenRequested() {
+    return dataLenRequested.get();
+  }
+
+  private void setDataLenRequested(long dataLenRequested) {
+    this.dataLenRequested.set(dataLenRequested);
+  }
+
+  private void updateDataLenRequested(long dataLenRequested){
+    this.dataLenRequested.addAndGet(dataLenRequested);
+  }
+
+  private boolean getCollectMetrics() {
+    return collectMetrics.get();
+  }
+
+  private void setCollectMetrics(boolean collectMetrics) {
+    this.collectMetrics.set(collectMetrics);
+  }
+
+  private boolean getIsParquetEvaluated() {
+    return isParquetEvaluated.get();
+  }
+
+  private void setIsParquetEvaluated(boolean isParquetEvaluated) {
+    this.isParquetEvaluated.set(isParquetEvaluated);
+  }
+
+  private boolean getIsLenUpdated() {
+    return isLenUpdated.get();
+  }
+
+  private void setIsLenUpdated(boolean isLenUpdated) {
+    this.isLenUpdated.set(isLenUpdated);
+  }
+
+  /**
+   * Updates the metrics map with an entry for the specified file if it doesn't already exist.
+   *
+   * @param filePathIdentifier The unique identifier for the file.
+   */
+  public void updateMap(String filePathIdentifier) {
+    // If the file is not already in the metrics map, add it with a new AbfsReadFooterMetrics object.
+    metricsMap.computeIfAbsent(filePathIdentifier, key -> new AbfsReadFooterMetrics());
+  }
+
+  /**
+   * Checks and updates metrics for a specific file identified by filePathIdentifier.
+   * If the metrics do not exist for the file, they are initialized.
+   *
+   * @param filePathIdentifier The unique identifier for the file.
+   * @param len               The length of the read operation.
+   * @param contentLength     The total content length of the file.
+   * @param nextReadPos       The position of the next read operation.
+   */
+  public void checkMetricUpdate(final String filePathIdentifier, final int len, final long contentLength,
+      final long nextReadPos) {
+    AbfsReadFooterMetrics readFooterMetrics = metricsMap.computeIfAbsent(
+            filePathIdentifier, key -> new AbfsReadFooterMetrics());
+    if (readFooterMetrics.getReadCount() == 0
+        || (readFooterMetrics.getReadCount() >= 1
+        && readFooterMetrics.getCollectMetrics())) {
+      updateMetrics(filePathIdentifier, len, contentLength, nextReadPos);
+    }
+  }
+
+  /**
+   * Updates metrics for a specific file identified by filePathIdentifier.
+   *
+   * @param filePathIdentifier  The unique identifier for the file.
+   * @param len                The length of the read operation.
+   * @param contentLength      The total content length of the file.
+   * @param nextReadPos        The position of the next read operation.
+   */
+  private void updateMetrics(final String filePathIdentifier, final int len, final long contentLength,
+                             final long nextReadPos) {
+    AbfsReadFooterMetrics readFooterMetrics = metricsMap.get(filePathIdentifier);
+
+    // Create a new AbfsReadFooterMetrics object if it doesn't exist in the metricsMap.
+    if (readFooterMetrics == null) {
+      readFooterMetrics = new AbfsReadFooterMetrics();
+      metricsMap.put(filePathIdentifier, readFooterMetrics);
+    }
+
+    int readCount;
+    synchronized (this) {
+      readCount = readFooterMetrics.incrementReadCount();
+    }
+
+    if (readCount == 1) {
+      // Update metrics for the first read.
+      updateMetricsOnFirstRead(readFooterMetrics, nextReadPos, len, contentLength);
+    }
+
+    synchronized (this) {
+      if (readFooterMetrics.getCollectLenMetrics()) {
+        readFooterMetrics.updateDataLenRequested(len);
+      }
+    }
+
+    if (readCount == 2) {
+      // Update metrics for the second read.
+      updateMetricsOnSecondRead(readFooterMetrics, nextReadPos, len);
+    }
+  }
+
+  /**
+   * Updates metrics for the first read operation.
+   *
+   * @param readFooterMetrics The metrics object to update.
+   * @param nextReadPos       The position of the next read operation.
+   * @param len               The length of the read operation.
+   * @param contentLength     The total content length of the file.
+   */
+  private void updateMetricsOnFirstRead(AbfsReadFooterMetrics readFooterMetrics, long nextReadPos, int len, long contentLength) {
+    if (nextReadPos >= contentLength - (long) Integer.parseInt(FOOTER_LENGTH) * ONE_KB) {
+      readFooterMetrics.setCollectMetrics(true);
+      readFooterMetrics.setCollectMetricsForNextRead(true);
+      readFooterMetrics.setOffsetOfFirstRead(nextReadPos);
+      readFooterMetrics.setSizeReadByFirstRead(len + "_" + Math.abs(contentLength - nextReadPos));
+      readFooterMetrics.setFileLength(contentLength);
+    }
+  }
+
+  /**
+   * Updates metrics for the second read operation.
+   *
+   * @param readFooterMetrics The metrics object to update.
+   * @param nextReadPos       The position of the next read operation.
+   * @param len               The length of the read operation.
+   */
+  private void updateMetricsOnSecondRead(AbfsReadFooterMetrics readFooterMetrics, long nextReadPos, int len) {
+    if (readFooterMetrics.getCollectMetricsForNextRead()) {
+      long offsetDiff = Math.abs(nextReadPos - readFooterMetrics.getOffsetOfFirstRead());
+      readFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(len + "_" + offsetDiff);
+      readFooterMetrics.setCollectLenMetrics(true);
+    }
+  }
+
+
+  /**
+   * Check if the given file should be marked as a Parquet file.
+   *
+   * @param metrics The metrics to evaluate.
+   * @return True if the file meet the criteria for being marked as a Parquet file, false otherwise.
+   */
+  private boolean shouldMarkAsParquet(AbfsReadFooterMetrics metrics) {
+    return metrics.getCollectMetrics()
+            && metrics.getReadCount() >= 2
+            && !metrics.getIsParquetEvaluated()
+            && haveEqualValues(metrics.getSizeReadByFirstRead())
+            && haveEqualValues(metrics.getOffsetDiffBetweenFirstAndSecondRead());
+  }
+
+  /**
+   * Check if two values are equal, considering they are in the format "value1_value2".
+   *
+   * @param value The value to check.
+   * @return True if the two parts of the value are equal, false otherwise.
+   */
+  private boolean haveEqualValues(String value) {
+    String[] parts = value.split("_");
+    return parts.length == 2 && parts[0].equals(parts[1]);
+  }
+
+  /**
+   * Mark the given metrics as a Parquet file and update related values.
+   *
+   * @param metrics The metrics to mark as Parquet.
+   */
+  private void markAsParquet(AbfsReadFooterMetrics metrics) {
+    metrics.setIsParquetFile(true);
+    String[] parts = metrics.getSizeReadByFirstRead().split("_");
+    metrics.setSizeReadByFirstRead(parts[0]);
+    parts = metrics.getOffsetDiffBetweenFirstAndSecondRead().split("_");
+    metrics.setOffsetDiffBetweenFirstAndSecondRead(parts[0]);
+    metrics.setIsParquetEvaluated(true);
+  }
+
+  /**
+   * Check each metric in the provided map and mark them as Parquet files if they meet the criteria.
+   *
+   * @param metricsMap The map containing metrics to evaluate.
+   */
+  public void checkIsParquet(Map<String, AbfsReadFooterMetrics> metricsMap) {
+    for (Map.Entry<String, AbfsReadFooterMetrics> entry : metricsMap.entrySet()) {
+      AbfsReadFooterMetrics readFooterMetrics = entry.getValue();
+      if (shouldMarkAsParquet(readFooterMetrics)) {
+        markAsParquet(readFooterMetrics);
+        metricsMap.replace(entry.getKey(), readFooterMetrics);
+      }
+    }
+  }
+
+  /**
+   * Updates the average read length requested for metrics of all files in the metrics map.
+   * If the metrics indicate that the update is needed, it calculates the average read length and updates the metrics.
+   *
+   * @param metricsMap A map containing metrics for different files with unique identifiers.
+   */
+  private void updateLenRequested(Map<String, AbfsReadFooterMetrics> metricsMap) {
+    for (AbfsReadFooterMetrics readFooterMetrics : metricsMap.values()) {
+      if (shouldUpdateLenRequested(readFooterMetrics)) {
+        int readReqCount = readFooterMetrics.getReadCount() - 2;
+        readFooterMetrics.setAvgReadLenRequested(
+                (double) readFooterMetrics.getDataLenRequested() / readReqCount);
+        readFooterMetrics.setIsLenUpdated(true);
+      }
+    }
+  }
+
+  /**
+   * Checks whether the average read length requested should be updated for the given metrics.
+   *
+   * The method returns true if the following conditions are met:
+   * - Metrics collection is enabled.
+   * - The number of read counts is greater than 2.
+   * - The average read length has not been updated previously.
+   *
+   * @param readFooterMetrics The metrics object to evaluate.
+   * @return True if the average read length should be updated, false otherwise.
+   */
+  private boolean shouldUpdateLenRequested(AbfsReadFooterMetrics readFooterMetrics) {
+    return readFooterMetrics.getCollectMetrics()
+            && readFooterMetrics.getReadCount() > 2
+            && !readFooterMetrics.getIsLenUpdated();
+  }
+
+  /**
+   * Calculates the average metrics from a list of AbfsReadFooterMetrics and sets the values in the provided 'avgParquetReadFooterMetrics' object.
+   *
+   * @param isParquetList The list of AbfsReadFooterMetrics to compute the averages from.
+   * @param avgParquetReadFooterMetrics The target AbfsReadFooterMetrics object to store the computed average values.
+   *
+   * This method calculates various average metrics from the provided list and sets them in the 'avgParquetReadFooterMetrics' object.
+   * The metrics include:
+   * - Size read by the first read
+   * - Offset difference between the first and second read
+   * - Average file length
+   * - Average requested read length
+   */
+  private void getParquetReadFooterMetricsAverage(List<AbfsReadFooterMetrics> isParquetList,
+      AbfsReadFooterMetrics avgParquetReadFooterMetrics){
+    avgParquetReadFooterMetrics.setSizeReadByFirstRead(
+        String.format("%.3f", isParquetList.stream()
+            .map(AbfsReadFooterMetrics::getSizeReadByFirstRead).mapToDouble(
+                Double::parseDouble).average().orElse(0.0)));
+    avgParquetReadFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(
+        String.format("%.3f", isParquetList.stream()
+            .map(AbfsReadFooterMetrics::getOffsetDiffBetweenFirstAndSecondRead)
+            .mapToDouble(Double::parseDouble).average().orElse(0.0)));
+    avgParquetReadFooterMetrics.setAvgFileLength(isParquetList.stream()
+        .mapToDouble(AbfsReadFooterMetrics::getFileLength).average().orElse(0.0));
+    avgParquetReadFooterMetrics.setAvgReadLenRequested(isParquetList.stream().
+        map(AbfsReadFooterMetrics::getAvgReadLenRequested).
+        mapToDouble(Double::doubleValue).average().orElse(0.0));
+  }
+
+  /**
+   * Calculates the average metrics from a list of non-Parquet AbfsReadFooterMetrics instances.
+   *
+   * This method takes a list of AbfsReadFooterMetrics representing non-Parquet reads and calculates
+   * the average values for the size read by the first read and the offset difference between the first
+   * and second read. The averages are then set in the provided AbfsReadFooterMetrics instance.
+   *
+   * @param isNonParquetList A list of AbfsReadFooterMetrics instances representing non-Parquet reads.
+   * @param avgNonParquetReadFooterMetrics The AbfsReadFooterMetrics instance to store the calculated averages.
+   *                                      It is assumed that the size of the list is at least 1, and the first
+   *                                      element of the list is used to determine the size of arrays.
+   *                                      The instance is modified in-place with the calculated averages.
+   *
+   *
+   **/
+  private void getNonParquetReadFooterMetricsAverage(List<AbfsReadFooterMetrics> isNonParquetList,
+                                                     AbfsReadFooterMetrics avgNonParquetReadFooterMetrics) {
+    int size = isNonParquetList.get(0).getSizeReadByFirstRead().split("_").length;
+    double[] store = new double[2 * size];
+    // Calculating sum of individual values
+    isNonParquetList.forEach(abfsReadFooterMetrics -> {
+      String[] firstReadSize = abfsReadFooterMetrics.getSizeReadByFirstRead().split("_");
+      String[] offDiffFirstSecondRead = abfsReadFooterMetrics.getOffsetDiffBetweenFirstAndSecondRead().split("_");
+
+      for (int i = 0; i < firstReadSize.length; i++) {
+        store[i] += Long.parseLong(firstReadSize[i]);
+        store[i + size] += Long.parseLong(offDiffFirstSecondRead[i]);
+      }
+    });
+
+    // Calculating averages and creating formatted strings
+    StringJoiner firstReadSize = new StringJoiner("_");
+    StringJoiner offDiffFirstSecondRead = new StringJoiner("_");
+
+    for (int j = 0; j < size; j++) {
+      firstReadSize.add(String.format("%.3f", store[j] / isNonParquetList.size()));
+      offDiffFirstSecondRead.add(String.format("%.3f", store[j + size] / isNonParquetList.size()));
+    }
+
+    avgNonParquetReadFooterMetrics.setSizeReadByFirstRead(firstReadSize.toString());
+    avgNonParquetReadFooterMetrics.setOffsetDiffBetweenFirstAndSecondRead(offDiffFirstSecondRead.toString());
+    avgNonParquetReadFooterMetrics.setAvgFileLength(isNonParquetList.stream()
+            .mapToDouble(AbfsReadFooterMetrics::getFileLength).average().orElse(0.0));
+    avgNonParquetReadFooterMetrics.setAvgReadLenRequested(isNonParquetList.stream()
+            .mapToDouble(AbfsReadFooterMetrics::getAvgReadLenRequested).average().orElse(0.0));
+  }
+
+  /*
+  Acronyms:
+  1.FR :- First Read (In case of parquet we only maintain the size requested by application for
+  the first read, in case of non parquet we maintain a string separated by "_" delimiter where the first
+  substring represents the len requested for first read and the second substring represents the seek pointer difference from the
+  end of the file.)
+  2.SR :- Second Read (In case of parquet we only maintain the size requested by application for
+  the second read, in case of non parquet we maintain a string separated by "_" delimiter where the first
+  substring represents the len requested for second read and the second substring represents the seek pointer difference from the
+  offset of the first read.)
+  3.FL :- Total length of the file requested for read
+   */
+  public String getReadFooterMetrics(AbfsReadFooterMetrics avgReadFooterMetrics) {
+    String readFooterMetric = "";
+    if (avgReadFooterMetrics.getIsParquetFile()) {
+      readFooterMetric += "$Parquet:";
+    } else {
+      readFooterMetric += "$NonParquet:";
+    }
+    readFooterMetric += "$FR=" + avgReadFooterMetrics.getSizeReadByFirstRead()
+        + "$SR="
+        + avgReadFooterMetrics.getOffsetDiffBetweenFirstAndSecondRead()
+        + "$FL=" + String.format("%.3f",
+        avgReadFooterMetrics.getAvgFileLength())
+        + "$RL=" + String.format("%.3f",
+        avgReadFooterMetrics.getAvgReadLenRequested());
+    return readFooterMetric;
+  }
+
+/**
+ * Retrieves and aggregates read footer metrics for both Parquet and non-Parquet files from a list
+ * of AbfsReadFooterMetrics instances. The function calculates the average metrics separately for
+ * Parquet and non-Parquet files and returns a formatted string containing the aggregated metrics.
+ *
+ * @param readFooterMetricsList A list of AbfsReadFooterMetrics instances containing read footer metrics
+ *                              for both Parquet and non-Parquet files.
+ *
+ * @return A formatted string containing the aggregated read footer metrics for both Parquet and non-Parquet files.
+ *
+ **/
+private String getFooterMetrics(List<AbfsReadFooterMetrics> readFooterMetricsList) {
+  List<AbfsReadFooterMetrics> isParquetList = new ArrayList<>();
+  List<AbfsReadFooterMetrics> isNonParquetList = new ArrayList<>();
+  for (AbfsReadFooterMetrics abfsReadFooterMetrics : readFooterMetricsList) {
+    if (abfsReadFooterMetrics.getIsParquetFile()) {
+      isParquetList.add(abfsReadFooterMetrics);
+    } else {
+      if (abfsReadFooterMetrics.getReadCount() >= 2) {

Review Comment:
   so it only goes into the non parquet file if its read the footer more than once?



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -1411,6 +1448,97 @@ protected AccessTokenProvider getTokenProvider() {
     return tokenProvider;
   }
 
+  public AzureBlobFileSystem getMetricFilesystem() throws IOException {
+    if (metricFs == null) {
+      try {
+        Configuration metricConfig = abfsConfiguration.getRawConfiguration();
+        String metricAccountKey = metricConfig.get(FS_AZURE_METRIC_ACCOUNT_KEY);
+        final String abfsMetricUrl = metricConfig.get(FS_AZURE_METRIC_URI);
+        if (abfsMetricUrl == null) {
+          return null;
+        }
+        metricConfig.set(FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME, metricAccountKey);
+        metricConfig.set(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION, "false");
+        URI metricUri;
+        metricUri = new URI(FileSystemUriSchemes.ABFS_SCHEME, abfsMetricUrl, null, null, null);
+        metricFs = (AzureBlobFileSystem) FileSystem.newInstance(metricUri, metricConfig);
+      } catch (AzureBlobFileSystemException | URISyntaxException ex) {
+        //do nothing
+      }
+    }
+    return metricFs;
+  }
+
+  private TracingContext getMetricTracingContext() {
+    String hostName;
+    try {
+      hostName = InetAddress.getLocalHost().getHostName();
+    } catch (UnknownHostException e) {
+      hostName = "UnknownHost";
+    }
+    return new TracingContext(TracingContext.validateClientCorrelationID(
+        abfsConfiguration.getClientCorrelationId()),
+        hostName, FSOperationType.GET_ATTR, true,
+        abfsConfiguration.getTracingHeaderFormat(),
+        null, abfsCounters.toString());
+  }
+
+  /**
+   * Synchronized method to suspend or resume timer.
+   * @param timerFunctionality resume or suspend.
+   * @param timerTask The timertask object.
+   * @return true or false.
+   */
+  synchronized boolean timerOrchestrator(TimerFunctionality timerFunctionality,
+      TimerTask timerTask) {
+    switch (timerFunctionality) {
+    case RESUME:
+      if (metricCollectionStopped.get()) {
+        resumeTimer();
+      }
+      break;
+    case SUSPEND:
+      long now = System.currentTimeMillis();
+      long lastExecutionTime = abfsCounters.getLastExecutionTime().get();
+      if (metricCollectionEnabled && (now - lastExecutionTime >= metricAnalysisPeriod)) {
+        timerTask.cancel();
+        timer.purge();
+        metricCollectionStopped.set(true);
+        return true;
+      }
+      break;
+    default:
+      break;
+    }
+    return false;
+  }
+
+  private void resumeTimer() {
+    metricCollectionStopped.set(false);
+    timer.schedule(new TimerTaskImpl(),
+        metricIdlePeriod,
+        metricIdlePeriod);
+  }
+
+  class TimerTaskImpl extends TimerTask {

Review Comment:
   this happens in close(), doesn't it?



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AzureBlobFileSystem.java:
##########
@@ -700,6 +698,18 @@ public synchronized void close() throws IOException {
     if (isClosed) {
       return;
     }
+    if (abfsStore.getClient().isMetricCollectionEnabled()) {
+      TracingContext tracingMetricContext = new TracingContext(
+              clientCorrelationId,
+              fileSystemId, FSOperationType.GET_ATTR, true,
+              tracingHeaderFormat,
+              listener, abfsCounters.toString());
+      try {
+        getAbfsClient().getMetricCall(tracingMetricContext);
+      } catch (IOException e) {
+        throw new IOException(e);

Review Comment:
   no



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -1655,7 +1694,7 @@ void setIsNamespaceEnabled(final Boolean isNamespaceEnabled) {
    * Getter for abfsCounters from AbfsClient.
    * @return AbfsCounters instance.
    */
-  protected AbfsCounters getAbfsCounters() {
+  public AbfsCounters getAbfsCounters() {

Review Comment:
   proposed: add an interface for the service methods and implement it; helps slightly to decouple things



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -388,26 +484,30 @@ private boolean executeHttpOperation(final int retryCount,
    */
   @VisibleForTesting
   public void signRequest(final AbfsHttpOperation httpOperation, int bytesToSign) throws IOException {
-    switch(client.getAuthType()) {
-      case Custom:
-      case OAuth:
-        LOG.debug("Authenticating request with OAuth2 access token");
-        httpOperation.getConnection().setRequestProperty(HttpHeaderConfigurations.AUTHORIZATION,
-            client.getAccessToken());
-        break;
-      case SAS:
-        // do nothing; the SAS token should already be appended to the query string
-        httpOperation.setMaskForSAS(); //mask sig/oid from url for logs
-        break;
-      case SharedKey:
-      default:
-        // sign the HTTP request
-        LOG.debug("Signing request with shared key");
-        // sign the HTTP request
-        client.getSharedKeyCredentials().signRequest(
-            httpOperation.getConnection(),
-            bytesToSign);
-        break;
+    if (client.isSendMetricCall()) {
+      client.getMetricSharedkeyCredentials().signRequest(httpOperation.getConnection(), bytesToSign);

Review Comment:
   log what is happening here



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -427,7 +527,7 @@ AbfsHttpOperation createHttpOperation() throws IOException {
   /**
    * Incrementing Abfs counters with a long value.
    *
-   * @param statistic the Abfs statistic that needs to be incremented.
+   * @param statistic the Abfs statistic that needs to be incremented.f

Review Comment:
   typo



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -436,6 +536,60 @@ private void incrementCounter(AbfsStatistic statistic, long value) {
     }
   }
 
+  /**
+   * Updates the count metrics based on the provided retry count.
+   * @param retryCount The retry count used to determine the metrics category.
+   *
+   * This method increments the number of succeeded requests for the specified retry count.
+   */
+  private void updateCount(int retryCount){
+      String retryCounter = getKey(retryCount);
+      metricsMap.get(retryCounter).incrementNumberOfRequestsSucceeded();
+  }
+
+  /**
+   * Updates backoff time metrics based on the provided retry count and sleep duration.
+   * @param retryCount    The retry count used to determine the metrics category.
+   * @param sleepDuration The duration of sleep during backoff.
+   *
+   * This method calculates and updates various backoff time metrics, including minimum, maximum,
+   * and total backoff time, as well as the total number of requests for the specified retry count.
+   */
+  private void updateBackoffTimeMetrics(int retryCount, long sleepDuration) {
+    synchronized (this) {
+      String retryCounter = getKey(retryCount);

Review Comment:
   again, IOStatistics does a lot of this, you'd have a duration which would set min/max and total+count, so the mean can be calculated too. 



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsRestOperation.java:
##########
@@ -436,6 +536,60 @@ private void incrementCounter(AbfsStatistic statistic, long value) {
     }
   }
 
+  /**
+   * Updates the count metrics based on the provided retry count.
+   * @param retryCount The retry count used to determine the metrics category.
+   *
+   * This method increments the number of succeeded requests for the specified retry count.
+   */
+  private void updateCount(int retryCount){
+      String retryCounter = getKey(retryCount);
+      metricsMap.get(retryCounter).incrementNumberOfRequestsSucceeded();
+  }
+
+  /**
+   * Updates backoff time metrics based on the provided retry count and sleep duration.
+   * @param retryCount    The retry count used to determine the metrics category.
+   * @param sleepDuration The duration of sleep during backoff.
+   *
+   * This method calculates and updates various backoff time metrics, including minimum, maximum,
+   * and total backoff time, as well as the total number of requests for the specified retry count.
+   */
+  private void updateBackoffTimeMetrics(int retryCount, long sleepDuration) {
+    synchronized (this) {
+      String retryCounter = getKey(retryCount);
+      AbfsBackoffMetrics abfsBackoffMetrics = metricsMap.get(retryCounter);
+      long minBackoffTime = Math.min(abfsBackoffMetrics.getMinBackoff(), sleepDuration);
+      long maxBackoffForTime = Math.max(abfsBackoffMetrics.getMaxBackoff(), sleepDuration);
+      long totalBackoffTime = abfsBackoffMetrics.getTotalBackoff() + sleepDuration;
+      abfsBackoffMetrics.incrementTotalRequests();
+      abfsBackoffMetrics.setMinBackoff(minBackoffTime);
+      abfsBackoffMetrics.setMaxBackoff(maxBackoffForTime);
+      abfsBackoffMetrics.setTotalBackoff(totalBackoffTime);
+      metricsMap.put(retryCounter, abfsBackoffMetrics);
+    }
+  }
+
+  /**
+   * Generates a key based on the provided retry count to categorize metrics.
+   *
+   * @param retryCount The retry count used to determine the key.
+   * @return A string key representing the metrics category for the given retry count.
+   *
+   * This method categorizes retry counts into different ranges and assigns a corresponding key.
+   */
+  private String getKey(int retryCount) {

Review Comment:
   and this needs a meaningful name



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsBackoffMetrics.java:
##########
@@ -0,0 +1,312 @@
+/**
+ * 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.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.Map;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.HUNDRED;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.THOUSAND;
+
+public class AbfsBackoffMetrics {
+
+  private AtomicLong numberOfRequestsSucceeded;
+
+  private AtomicLong minBackoff;
+
+  private AtomicLong maxBackoff;
+
+  private AtomicLong totalRequests;
+
+  private AtomicLong totalBackoff;
+
+  private String retryCount;
+
+  private AtomicLong numberOfIOPSThrottledRequests;
+
+  private AtomicLong numberOfBandwidthThrottledRequests;
+
+  private AtomicLong numberOfOtherThrottledRequests;
+
+  private AtomicLong numberOfNetworkFailedRequests;
+
+  private AtomicLong maxRetryCount;
+
+  private AtomicLong totalNumberOfRequests;
+
+  private AtomicLong numberOfRequestsSucceededWithoutRetrying;
+
+  private AtomicLong numberOfRequestsFailed;
+
+  private final Map<String, AbfsBackoffMetrics> metricsMap
+      = new ConcurrentHashMap<>();
+
+  public AbfsBackoffMetrics() {
+    initializeMap();

Review Comment:
   I'd suggest you actually use an {{IOStatisticsStoreBuilder}} to build an IOStatisticsStore, from which you can then extract the counters for direct access -but still be able to snapshot, aggregate and share the stats though public APIs.
   
   See `S3AInstrumentation.InputStreamStatistics` as an example of this



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java:
##########
@@ -41,6 +41,10 @@ public final class ConfigurationKeys {
    */
   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_METRIC_ACCOUNT_NAME = "fs.azure.metric.account.name";

Review Comment:
   add javadocs for all of these. 



##########
hadoop-tools/hadoop-azure/src/site/markdown/abfs.md:
##########
@@ -1052,6 +1052,49 @@ Note that these performance numbers are also sent back to the ADLS Gen 2 API end
 in the `x-ms-abfs-client-latency` HTTP headers in subsequent requests. Azure uses these
 settings to track their end-to-end latency.
 
+### <a name="drivermetricoptions"></a> Driver Metric Options
+

Review Comment:
   add a paragraph explaining what and why



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/utils/MetricFormat.java:
##########
@@ -0,0 +1,36 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs.utils;
+
+public enum MetricFormat {
+  INTERNAL_BACKOFF_METRIC_FORMAT, // <client-correlation-id>:<client-req-id>:<filesystem-id>

Review Comment:
   move these to javadocs, taking care to use &lt; to escape the < properly



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -22,21 +22,28 @@
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.net.HttpURLConnection;
+import java.net.InetAddress;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLEncoder;
+import java.net.UnknownHostException;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
 import java.util.Base64;
 import java.util.List;
 import java.util.Locale;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.UUID;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;

Review Comment:
   below line55. and remember that because I do always pick up on these, you can save time by reviewing yourself



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";
+
+  /**
+   * Integration test for reading footer metrics with both Parquet and non-Parquet reads.
+   */
+  @Test
+  public void testReadFooterMetricsWithParquetAndNonParquet() throws Exception {
+    testReadWriteAndSeek(8 * ONE_MB, DEFAULT_READ_BUFFER_SIZE, ONE_KB, 4 * ONE_KB);
+  }
+
+  /**
+   * Configures the AzureBlobFileSystem with the given buffer size.
+   *
+   * @param bufferSize Buffer size to set for write and read operations.
+   * @return AbfsConfiguration used for configuration.
+   */
+  private Configuration getConfiguration(int bufferSize) {
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_FOOTER_METRIC_FORMAT));
+    configuration.setInt(AZURE_READ_BUFFER_SIZE, bufferSize);
+    configuration.setInt(AZURE_WRITE_BUFFER_SIZE, bufferSize);
+    return configuration;
+  }
+
+  /**
+   * Writes data to the specified file path in the AzureBlobFileSystem.
+   *
+   * @param fs      AzureBlobFileSystem instance.
+   * @param testPath Path to the file.
+   * @param data    Data to write to the file.
+   */
+  private void writeDataToFile(AzureBlobFileSystem fs, Path testPath, byte[] data) throws IOException {
+    FSDataOutputStream stream = fs.create(testPath);

Review Comment:
   use try with resources



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";
+
+  /**
+   * Integration test for reading footer metrics with both Parquet and non-Parquet reads.
+   */
+  @Test
+  public void testReadFooterMetricsWithParquetAndNonParquet() throws Exception {
+    testReadWriteAndSeek(8 * ONE_MB, DEFAULT_READ_BUFFER_SIZE, ONE_KB, 4 * ONE_KB);
+  }
+
+  /**
+   * Configures the AzureBlobFileSystem with the given buffer size.
+   *
+   * @param bufferSize Buffer size to set for write and read operations.
+   * @return AbfsConfiguration used for configuration.
+   */
+  private Configuration getConfiguration(int bufferSize) {
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_FOOTER_METRIC_FORMAT));
+    configuration.setInt(AZURE_READ_BUFFER_SIZE, bufferSize);
+    configuration.setInt(AZURE_WRITE_BUFFER_SIZE, bufferSize);
+    return configuration;
+  }
+
+  /**
+   * Writes data to the specified file path in the AzureBlobFileSystem.
+   *
+   * @param fs      AzureBlobFileSystem instance.
+   * @param testPath Path to the file.
+   * @param data    Data to write to the file.
+   */
+  private void writeDataToFile(AzureBlobFileSystem fs, Path testPath, byte[] data) throws IOException {
+    FSDataOutputStream stream = fs.create(testPath);
+    try {
+      stream.write(data);
+    } finally {
+      stream.close();
+    }
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG, IOSTATISTICS_LOGGING_LEVEL_INFO, stream);
+  }
+
+  /**
+   * Asserts that the actual metrics obtained from the AzureBlobFileSystem match the expected metrics string.
+   *
+   * @param fs               AzureBlobFileSystem instance.
+   * @param expectedMetrics  Expected metrics string.
+   */
+  private void assertMetricsEquality(AzureBlobFileSystem fs, String expectedMetrics) {
+    AbfsReadFooterMetrics actualMetrics = fs.getAbfsClient().getAbfsCounters().getAbfsReadFooterMetrics();
+    assertNotNull("AbfsReadFooterMetrics is null", actualMetrics);
+    assertEquals("The computed metrics differs from the actual metrics", expectedMetrics, actualMetrics.toString());
+  }
+
+  /**
+   * Test for reading footer metrics with a non-Parquet file.
+   */
+  @Test
+  public void testReadFooterMetrics() throws Exception {
+    // Initialize AzureBlobFileSystem and set buffer size for configuration.
+    int bufferSize = MIN_BUFFER_SIZE;
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);

Review Comment:
   needs to be closed at end of test case



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/contracts/services/AzureServiceErrorCode.java:
##########
@@ -21,7 +21,8 @@
 import java.net.HttpURLConnection;
 import java.util.ArrayList;
 import java.util.List;
-
+import org.slf4j.Logger;

Review Comment:
   self contained block, separate from the others



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -181,6 +198,35 @@ private AbfsClient(final URL baseUrl,
         new ThreadFactoryBuilder().setNameFormat("AbfsClient Lease Ops").setDaemon(true).build();
     this.executorService = MoreExecutors.listeningDecorator(
         HadoopExecutors.newScheduledThreadPool(this.abfsConfiguration.getNumLeaseThreads(), tf));
+    this.metricFormat = abfsConfiguration.getMetricFormat();
+    this.isMetricCollectionStopped = new AtomicBoolean(false);
+    this.metricAnalysisPeriod = abfsConfiguration.getMetricAnalysisTimeout();
+    this.metricIdlePeriod = abfsConfiguration.getMetricIdleTimeout();
+    if (!metricFormat.toString().equals("")) {
+      isMetricCollectionEnabled = true;

Review Comment:
   log at debug that metrics are being set up, including account name and format.



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";
+
+  /**
+   * Integration test for reading footer metrics with both Parquet and non-Parquet reads.
+   */
+  @Test
+  public void testReadFooterMetricsWithParquetAndNonParquet() throws Exception {
+    testReadWriteAndSeek(8 * ONE_MB, DEFAULT_READ_BUFFER_SIZE, ONE_KB, 4 * ONE_KB);
+  }
+
+  /**
+   * Configures the AzureBlobFileSystem with the given buffer size.
+   *
+   * @param bufferSize Buffer size to set for write and read operations.
+   * @return AbfsConfiguration used for configuration.
+   */
+  private Configuration getConfiguration(int bufferSize) {
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_FOOTER_METRIC_FORMAT));
+    configuration.setInt(AZURE_READ_BUFFER_SIZE, bufferSize);
+    configuration.setInt(AZURE_WRITE_BUFFER_SIZE, bufferSize);
+    return configuration;
+  }
+
+  /**
+   * Writes data to the specified file path in the AzureBlobFileSystem.
+   *
+   * @param fs      AzureBlobFileSystem instance.
+   * @param testPath Path to the file.
+   * @param data    Data to write to the file.
+   */
+  private void writeDataToFile(AzureBlobFileSystem fs, Path testPath, byte[] data) throws IOException {
+    FSDataOutputStream stream = fs.create(testPath);
+    try {
+      stream.write(data);
+    } finally {
+      stream.close();
+    }
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG, IOSTATISTICS_LOGGING_LEVEL_INFO, stream);
+  }
+
+  /**
+   * Asserts that the actual metrics obtained from the AzureBlobFileSystem match the expected metrics string.
+   *
+   * @param fs               AzureBlobFileSystem instance.
+   * @param expectedMetrics  Expected metrics string.
+   */
+  private void assertMetricsEquality(AzureBlobFileSystem fs, String expectedMetrics) {
+    AbfsReadFooterMetrics actualMetrics = fs.getAbfsClient().getAbfsCounters().getAbfsReadFooterMetrics();
+    assertNotNull("AbfsReadFooterMetrics is null", actualMetrics);
+    assertEquals("The computed metrics differs from the actual metrics", expectedMetrics, actualMetrics.toString());
+  }
+
+  /**
+   * Test for reading footer metrics with a non-Parquet file.
+   */
+  @Test
+  public void testReadFooterMetrics() throws Exception {
+    // Initialize AzureBlobFileSystem and set buffer size for configuration.
+    int bufferSize = MIN_BUFFER_SIZE;
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);

Review Comment:
   needs to be closed at end of test case in a try/finally clause. if a test fails, we stil need that close()



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -181,6 +198,35 @@ private AbfsClient(final URL baseUrl,
         new ThreadFactoryBuilder().setNameFormat("AbfsClient Lease Ops").setDaemon(true).build();
     this.executorService = MoreExecutors.listeningDecorator(
         HadoopExecutors.newScheduledThreadPool(this.abfsConfiguration.getNumLeaseThreads(), tf));
+    this.metricFormat = abfsConfiguration.getMetricFormat();
+    this.isMetricCollectionStopped = new AtomicBoolean(false);
+    this.metricAnalysisPeriod = abfsConfiguration.getMetricAnalysisTimeout();
+    this.metricIdlePeriod = abfsConfiguration.getMetricIdleTimeout();
+    if (!metricFormat.toString().equals("")) {
+      isMetricCollectionEnabled = true;
+      abfsCounters.initializeMetrics(metricFormat);
+      String metricAccountName = abfsConfiguration.getMetricAccount();
+      int dotIndex = metricAccountName.indexOf(AbfsHttpConstants.DOT);
+      if (dotIndex <= 0) {
+        throw new InvalidUriException(
+                metricAccountName + " - account name is not fully qualified.");
+      }
+      String metricAccountKey = abfsConfiguration.getMetricAccountKey();
+      try {
+        metricSharedkeyCredentials = new SharedKeyCredentials(metricAccountName.substring(0, dotIndex),
+                metricAccountKey);
+      } catch (IllegalArgumentException e) {
+        throw new IOException("Exception while initializing metric credentials " + e);

Review Comment:
   stack traces are too precious to lose, include in the constructor



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/AbfsCountersImpl.java:
##########
@@ -63,6 +66,12 @@ public class AbfsCountersImpl implements AbfsCounters {
 
   private final IOStatisticsStore ioStatisticsStore;
 
+  private AbfsBackoffMetrics abfsBackoffMetrics = null;
+
+  private AbfsReadFooterMetrics abfsReadFooterMetrics = null;
+
+  private AtomicLong lastExecutionTime = null;

Review Comment:
   this should be final and constructed here



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";

Review Comment:
   that's a long sleep, which is going to hurt overall test performance.
   
   1. use something smaller
   2. use _ between groups of three
   3. or better, a Duration instance -which removes ambiguity about unit. If not: include MILLIS in the name



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAzureBlobFileSystemListStatus.java:
##########
@@ -80,8 +80,8 @@ public ITestAzureBlobFileSystemListStatus() throws Exception {
   public void testListPath() throws Exception {
     Configuration config = new Configuration(this.getRawConfiguration());
     config.set(AZURE_LIST_MAX_RESULTS, "5000");
-    try (final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem
-        .newInstance(getFileSystem().getUri(), config)) {
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem

Review Comment:
   try/finally



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsClient.java:
##########
@@ -1731,6 +1792,128 @@ protected AccessTokenProvider getTokenProvider() {
     return tokenProvider;
   }
 
+  /**
+   * Retrieves a TracingContext object configured for metric tracking.
+   * This method creates a TracingContext object with the validated client correlation ID,
+   * the host name of the local machine (or "UnknownHost" if unable to determine),
+   * the file system operation type set to GET_ATTR, and additional configuration parameters
+   * for metric tracking.
+   * The TracingContext is intended for use in tracking metrics related to Azure Blob FileSystem (ABFS) operations.
+   *
+   * @return A TracingContext object configured for metric tracking.
+   */
+  private TracingContext getMetricTracingContext() {
+    String hostName;
+    try {
+      hostName = InetAddress.getLocalHost().getHostName();

Review Comment:
   you should cache this



##########
hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/ITestAbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,385 @@
+
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs;
+
+import static org.apache.hadoop.fs.CommonConfigurationKeys.IOSTATISTICS_LOGGING_LEVEL_INFO;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_READ_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_WRITE_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_METRIC_FORMAT;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.MIN_BUFFER_SIZE;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.DEFAULT_READ_BUFFER_SIZE;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.azurebfs.utils.MetricFormat;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Random;
+import org.apache.hadoop.fs.azurebfs.services.AbfsReadFooterMetrics;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.statistics.IOStatisticsLogging;
+import org.apache.hadoop.fs.statistics.IOStatisticsSource;
+import org.apache.hadoop.fs.azurebfs.services.AbfsInputStream;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.azurebfs.utils.TracingHeaderValidator;
+import org.apache.hadoop.fs.azurebfs.constants.FSOperationType;
+
+public class ITestAbfsReadFooterMetrics extends AbstractAbfsScaleTest {
+
+  public ITestAbfsReadFooterMetrics() throws Exception {
+  }
+
+  private static final String TEST_PATH = "/testfile";
+  private static final String SLEEP_PERIOD = "90000";
+
+  /**
+   * Integration test for reading footer metrics with both Parquet and non-Parquet reads.
+   */
+  @Test
+  public void testReadFooterMetricsWithParquetAndNonParquet() throws Exception {
+    testReadWriteAndSeek(8 * ONE_MB, DEFAULT_READ_BUFFER_SIZE, ONE_KB, 4 * ONE_KB);
+  }
+
+  /**
+   * Configures the AzureBlobFileSystem with the given buffer size.
+   *
+   * @param bufferSize Buffer size to set for write and read operations.
+   * @return AbfsConfiguration used for configuration.
+   */
+  private Configuration getConfiguration(int bufferSize) {
+    final Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_METRIC_FORMAT, String.valueOf(MetricFormat.INTERNAL_FOOTER_METRIC_FORMAT));
+    configuration.setInt(AZURE_READ_BUFFER_SIZE, bufferSize);
+    configuration.setInt(AZURE_WRITE_BUFFER_SIZE, bufferSize);
+    return configuration;
+  }
+
+  /**
+   * Writes data to the specified file path in the AzureBlobFileSystem.
+   *
+   * @param fs      AzureBlobFileSystem instance.
+   * @param testPath Path to the file.
+   * @param data    Data to write to the file.
+   */
+  private void writeDataToFile(AzureBlobFileSystem fs, Path testPath, byte[] data) throws IOException {
+    FSDataOutputStream stream = fs.create(testPath);
+    try {
+      stream.write(data);
+    } finally {
+      stream.close();
+    }
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG, IOSTATISTICS_LOGGING_LEVEL_INFO, stream);
+  }
+
+  /**
+   * Asserts that the actual metrics obtained from the AzureBlobFileSystem match the expected metrics string.
+   *
+   * @param fs               AzureBlobFileSystem instance.
+   * @param expectedMetrics  Expected metrics string.
+   */
+  private void assertMetricsEquality(AzureBlobFileSystem fs, String expectedMetrics) {
+    AbfsReadFooterMetrics actualMetrics = fs.getAbfsClient().getAbfsCounters().getAbfsReadFooterMetrics();
+    assertNotNull("AbfsReadFooterMetrics is null", actualMetrics);
+    assertEquals("The computed metrics differs from the actual metrics", expectedMetrics, actualMetrics.toString());
+  }
+
+  /**
+   * Test for reading footer metrics with a non-Parquet file.
+   */
+  @Test
+  public void testReadFooterMetrics() throws Exception {
+    // Initialize AzureBlobFileSystem and set buffer size for configuration.
+    int bufferSize = MIN_BUFFER_SIZE;
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
+    AbfsConfiguration abfsConfiguration = fs.getAbfsStore().getAbfsConfiguration();
+
+    // Generate random data to write to the test file.
+    final byte[] b = new byte[2 * bufferSize];
+    new Random().nextBytes(b);
+
+    // Set up the test file path.
+    Path testPath = path(TEST_PATH);
+
+    // Write random data to the test file.
+    writeDataToFile(fs, testPath, b);
+
+    // Initialize a buffer for reading data.
+    final byte[] readBuffer = new byte[2 * bufferSize];
+    int result;
+
+    // Initialize statistics source for logging.
+    IOStatisticsSource statisticsSource = null;
+
+    try (FSDataInputStream inputStream = fs.open(testPath)) {
+      // Register a listener for tracing header validation.
+      statisticsSource = inputStream;
+      ((AbfsInputStream) inputStream.getWrappedStream()).registerListener(
+              new TracingHeaderValidator(abfsConfiguration.getClientCorrelationId(),
+                      fs.getFileSystemId(), FSOperationType.READ, true, 0,
+                      ((AbfsInputStream) inputStream.getWrappedStream())
+                              .getStreamID()));
+
+      // Perform the first read operation with seek.
+      inputStream.seek(bufferSize);
+      result = inputStream.read(readBuffer, bufferSize, bufferSize);
+      assertNotEquals(-1, result);
+
+      // To test tracingHeader for case with bypassReadAhead == true
+      inputStream.seek(0);
+      byte[] temp = new byte[5];
+      int t = inputStream.read(temp, 0, 1);
+
+      // Seek back to the beginning and perform another read operation.
+      inputStream.seek(0);
+      result = inputStream.read(readBuffer, 0, bufferSize);
+    }
+
+    // Log IO statistics at the INFO level.
+    IOStatisticsLogging.logIOStatisticsAtLevel(LOG,
+            IOSTATISTICS_LOGGING_LEVEL_INFO, statisticsSource);
+
+    // Ensure data is read successfully and matches the written data.
+    assertNotEquals("data read in final read()", -1, result);
+    assertArrayEquals(readBuffer, b);
+
+    // Get non-Parquet metrics and assert metrics equality.
+    AbfsReadFooterMetrics nonParquetMetrics = getNonParquetMetrics();
+    String metrics = nonParquetMetrics.getReadFooterMetrics(nonParquetMetrics);
+    assertMetricsEquality(fs, metrics);
+
+    // Close the AzureBlobFileSystem.
+    fs.close();
+  }
+
+  /**
+   * Generates and returns an instance of AbfsReadFooterMetrics for non-Parquet files.
+   */
+  private AbfsReadFooterMetrics getNonParquetMetrics() {
+    AbfsReadFooterMetrics nonParquetMetrics = new AbfsReadFooterMetrics();
+    nonParquetMetrics.setIsParquetFile(false);
+    nonParquetMetrics.setSizeReadByFirstRead("16384.000_16384.000");
+    nonParquetMetrics.setOffsetDiffBetweenFirstAndSecondRead("1.000_16384.000");
+    nonParquetMetrics.setAvgFileLength(Double.parseDouble("32768.000"));
+    nonParquetMetrics.setAvgReadLenRequested(Double.parseDouble("16384.000"));
+    return nonParquetMetrics;
+  }
+
+  /**
+   * Generates and returns an instance of AbfsReadFooterMetrics for parquet files.
+   */
+  private AbfsReadFooterMetrics getParquetMetrics() {
+    AbfsReadFooterMetrics parquetMetrics = new AbfsReadFooterMetrics();
+    parquetMetrics.setIsParquetFile(true);
+    parquetMetrics.setSizeReadByFirstRead("1024.000");
+    parquetMetrics.setOffsetDiffBetweenFirstAndSecondRead("4096.000");
+    parquetMetrics.setAvgFileLength(Double.parseDouble("8388608.000"));
+    parquetMetrics.setAvgReadLenRequested(0.000);
+    return parquetMetrics;
+  }
+
+  /**
+   * Test for reading, writing, and seeking with footer metrics.
+   *
+   * This method performs the integration test for reading, writing, and seeking operations
+   * with footer metrics. It creates an AzureBlobFileSystem, configures it, writes random data
+   * to a test file, performs read and seek operations, and checks the footer metrics for both
+   * Parquet and non-Parquet scenarios.
+   *
+   * @param fileSize Size of the test file.
+   * @param bufferSize Size of the buffer used for read and write operations.
+   * @param seek1 The position to seek to in the test file.
+   * @param seek2 Additional position to seek to in the test file (if not 0).
+   */
+  private void testReadWriteAndSeek(int fileSize, int bufferSize, Integer seek1, Integer seek2) throws Exception {
+    // Create an AzureBlobFileSystem instance.
+    Configuration configuration = getConfiguration(bufferSize);
+    final AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(configuration);
+    AbfsConfiguration abfsConfiguration = fs.getAbfsStore().getAbfsConfiguration();
+
+    // Generate random data to write to the test file.
+    final byte[] b = new byte[fileSize];
+    new Random().nextBytes(b);
+
+    // Define the path for the test file.
+    Path testPath = path("/testfile");
+
+    // Write the random data to the test file.
+    writeDataToFile(fs, testPath, b);
+
+    // Initialize a buffer for reading.
+    final byte[] readBuffer = new byte[fileSize];
+
+    // Initialize a source for IO statistics.
+    IOStatisticsSource statisticsSource = null;
+
+    // Open an input stream for the test file.
+    FSDataInputStream inputStream = fs.open(testPath);
+    statisticsSource = inputStream;
+
+    // Register a listener for tracing headers.
+    ((AbfsInputStream) inputStream.getWrappedStream()).registerListener(
+            new TracingHeaderValidator(abfsConfiguration.getClientCorrelationId(),
+                    fs.getFileSystemId(), FSOperationType.READ, true, 0,
+                    ((AbfsInputStream) inputStream.getWrappedStream())
+                            .getStreamID()));
+
+    // Seek to the specified position in the test file and read data.
+    inputStream.seek(fileSize - seek1);
+    inputStream.read(readBuffer, 0, seek1);
+
+    // If seek2 is non-zero, perform an additional seek and read.
+    if (seek2 != 0) {
+      inputStream.seek(fileSize - seek1 - seek2);
+      inputStream.read(readBuffer, 0, seek2);
+    }
+
+    // Close the input stream.
+    inputStream.close();
+
+    // Set a new buffer size for read and write operations.
+    int bufferSize1 = MIN_BUFFER_SIZE;
+    abfsConfiguration.setWriteBufferSize(bufferSize1);
+    abfsConfiguration.setReadBufferSize(bufferSize1);
+
+    // Generate new random data for a second test file.
+    final byte[] b1 = new byte[2 * bufferSize1];
+    new Random().nextBytes(b1);
+
+    // Define the path for the second test file.
+    Path testPath1 = path("/testfile1");
+
+    // Write the new random data to the second test file.
+    writeDataToFile(fs, testPath1, b1);
+
+    // Initialize a buffer for reading from the second test file.
+    final byte[] readBuffer1 = new byte[2 * bufferSize1];
+
+    // Open an input stream for the second test file.
+    FSDataInputStream inputStream1 = fs.open(testPath1);
+    statisticsSource = inputStream1;
+
+    // Register a listener for tracing headers.
+    ((AbfsInputStream) inputStream1.getWrappedStream()).registerListener(
+            new TracingHeaderValidator(abfsConfiguration.getClientCorrelationId(),
+                    fs.getFileSystemId(), FSOperationType.READ, true, 0,
+                    ((AbfsInputStream) inputStream1.getWrappedStream())
+                            .getStreamID()));
+
+    // Seek to a position in the second test file and read data.
+    inputStream1.seek(bufferSize1);
+    inputStream1.read(readBuffer1, bufferSize1, bufferSize1);
+
+    // To test tracingHeader for case with bypassReadAhead == true.
+    inputStream1.seek(0);
+    byte[] temp = new byte[5];
+    int t = inputStream1.read(temp, 0, 1);
+
+    // Seek to the beginning of the second test file and read data.
+    inputStream1.seek(0);
+    inputStream1.read(readBuffer1, 0, bufferSize1);
+
+    // Close the input stream for the second test file.
+    inputStream1.close();

Review Comment:
   again, try/finally



##########
hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsReadFooterMetrics.java:
##########
@@ -0,0 +1,549 @@
+/**
+ * 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.
+ */
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.util.Map;
+import java.util.StringJoiner;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.List;
+import java.util.ArrayList;
+
+import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
+
+public class AbfsReadFooterMetrics {

Review Comment:
   again, I think a lot of the numbers could be collected in IOStatisticsStore so easily printed and marshalled



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: common-issues-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-issues-help@hadoop.apache.org