IoStats-CallBack-AAL (#298)

## Description of change
<!-- Thank you for submitting a pull request!-->
<!-- Please describe your contribution here. What and why? -->
<!-- Please ensure your commit messages follow these
[guidelines](https://chris.beams.io/posts/git-commit/). -->
This PR implements a callback mechanism that allows the analytics
accelerator to pass metrics back to the S3A connector, ensuring that
each GET operation properly increments the
STREAM_READ_ANALYTICS_GET_REQUESTS counter. This enables S3A to monitor
and assert with accurate statistics.

Public Jira: https://issues.apache.org/jira/browse/HADOOP-19364

Haddop S3A PR: apache/hadoop#7763


---

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license and I agree to the terms of
the [Developer Certificate of Origin
(DCO)](https://developercertificate.org/).

Author: vaibhav5140

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/DefaultRequestCallbackImpl.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed 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 software.amazon.s3.analyticsaccelerator.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Default implementation of {@link RequestCallback} that logs requests at debug level. */
+public class DefaultRequestCallbackImpl implements RequestCallback {
+  private static final Logger LOG = LoggerFactory.getLogger(DefaultRequestCallbackImpl.class);
+
+  @Override
+  public void onGetRequest() {
+    LOG.trace("GET request made");
+  }
+
+  @Override
+  public void onHeadRequest() {
+    LOG.trace("HEAD request made");
+  }
+}

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java

@@ -42,6 +42,7 @@ public class OpenStreamInformation {
   private final StreamAuditContext streamAuditContext;
   private final ObjectMetadata objectMetadata;
   private final InputPolicy inputPolicy;
+  @Builder.Default private RequestCallback requestCallback = new DefaultRequestCallbackImpl();
   private final EncryptionSecrets encryptionSecrets;
 
   /** Default set of settings for {@link OpenStreamInformation} */

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/RequestCallback.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed 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 software.amazon.s3.analyticsaccelerator.util;
+/**
+ * Callback interface for S3 operations in Analytics Accelerator Library. Provides methods to track
+ * GET and HEAD requests.
+ */
+public interface RequestCallback {
+  /** Called when a GET request is made. */
+  void onGetRequest();
+  /** Called when a HEAD request is made. */
+  void onHeadRequest();
+}

common/src/test/java/software/amazon/s3/analyticsaccelerator/util/DefaultRequestCallbackImplTest.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed 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 software.amazon.s3.analyticsaccelerator.util;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+import org.junit.jupiter.api.Test;
+
+public class DefaultRequestCallbackImplTest {
+  @Test
+  public void testCallbackMethodsDoNotThrow() {
+    DefaultRequestCallbackImpl callback = new DefaultRequestCallbackImpl();
+
+    // No exceptions
+    assertDoesNotThrow(() -> callback.onGetRequest());
+    assertDoesNotThrow(() -> callback.onHeadRequest());
+  }
+}

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/S3SeekableInputStreamFactory.java

@@ -80,7 +80,8 @@ public S3SeekableInputStreamFactory(
     this.parquetColumnPrefetchStore =
         new ParquetColumnPrefetchStore(configuration.getLogicalIOConfiguration());
     this.objectMetadataStore =
-        new MetadataStore(objectClient, telemetry, configuration.getPhysicalIOConfiguration());
+        new MetadataStore(
+            objectClient, telemetry, configuration.getPhysicalIOConfiguration(), metrics);
     this.objectFormatSelector = new ObjectFormatSelector(configuration.getLogicalIOConfiguration());
     this.objectBlobStore =
         new BlobStore(objectClient, telemetry, configuration.getPhysicalIOConfiguration(), metrics);

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/Block.java

@@ -127,6 +127,8 @@ private void generateSourceAndData() throws IOException {
                 .referrer(referrer)
                 .build();
 
+        openStreamInformation.getRequestCallback().onGetRequest();
+
         this.source =
             this.telemetry.measureCritical(
                 () ->

input-stream/src/main/java/software/amazon/s3/analyticsaccelerator/io/physical/data/MetadataStore.java

@@ -25,12 +25,14 @@
 import lombok.NonNull;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.s3.analyticsaccelerator.common.Metrics;
 import software.amazon.s3.analyticsaccelerator.common.telemetry.Operation;
 import software.amazon.s3.analyticsaccelerator.common.telemetry.Telemetry;
 import software.amazon.s3.analyticsaccelerator.io.physical.PhysicalIOConfiguration;
 import software.amazon.s3.analyticsaccelerator.request.HeadRequest;
 import software.amazon.s3.analyticsaccelerator.request.ObjectClient;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
+import software.amazon.s3.analyticsaccelerator.util.MetricKey;
 import software.amazon.s3.analyticsaccelerator.util.OpenStreamInformation;
 import software.amazon.s3.analyticsaccelerator.util.S3URI;
 import software.amazon.s3.analyticsaccelerator.util.StreamAttributes;
@@ -45,6 +47,7 @@ public class MetadataStore implements Closeable {
   private final Telemetry telemetry;
   private final Map<S3URI, CompletableFuture<ObjectMetadata>> cache;
   private final PhysicalIOConfiguration configuration;
+  private final Metrics aggregatingMetrics;
 
   private static final Logger LOG = LoggerFactory.getLogger(MetadataStore.class);
   private static final String OPERATION_METADATA_HEAD_ASYNC = "metadata.store.head.async";
@@ -55,14 +58,17 @@ public class MetadataStore implements Closeable {
    *
    * @param objectClient the object client to use for object store interactions.
    * @param telemetry The {@link Telemetry} to use to report measurements.
+   * @param aggregatingMetrics blobstore metrics
    * @param configuration a configuration of PhysicalIO.
    */
   public MetadataStore(
       @NonNull ObjectClient objectClient,
       @NonNull Telemetry telemetry,
-      @NonNull PhysicalIOConfiguration configuration) {
+      @NonNull PhysicalIOConfiguration configuration,
+      @NonNull Metrics aggregatingMetrics) {
     this.objectClient = objectClient;
     this.telemetry = telemetry;
+    this.aggregatingMetrics = aggregatingMetrics;
     this.cache =
         Collections.synchronizedMap(
             new LinkedHashMap<S3URI, CompletableFuture<ObjectMetadata>>() {
@@ -125,8 +131,14 @@ public synchronized CompletableFuture<ObjectMetadata> asyncGet(
                         .name(OPERATION_METADATA_HEAD_ASYNC)
                         .attribute(StreamAttributes.uri(s3URI))
                         .build(),
-                objectClient.headObject(
-                    HeadRequest.builder().s3Uri(s3URI).build(), openStreamInformation)));
+                () -> {
+                  CompletableFuture<ObjectMetadata> result =
+                      objectClient.headObject(
+                          HeadRequest.builder().s3Uri(s3URI).build(), openStreamInformation);
+                  openStreamInformation.getRequestCallback().onHeadRequest();
+                  this.aggregatingMetrics.add(MetricKey.HEAD_REQUEST_COUNT, 1);
+                  return result;
+                }));
   }
 
   /**

input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/S3SeekableInputStreamFactoryTest.java

@@ -102,7 +102,7 @@ void testCreateDefaultStream() throws IOException {
 
     inputStream =
         s3SeekableInputStreamFactory.createStream(
-            S3URI.of("bucket", "key"), mock(OpenStreamInformation.class));
+            S3URI.of("bucket", "key"), OpenStreamInformation.DEFAULT);
     assertNotNull(inputStream);
   }
 
@@ -165,8 +165,7 @@ void testCreateIndependentStream() throws IOException {
     S3SeekableInputStream inputStream = s3SeekableInputStreamFactory.createStream(s3URI);
     assertNotNull(inputStream);
 
-    inputStream =
-        s3SeekableInputStreamFactory.createStream(s3URI, mock(OpenStreamInformation.class));
+    inputStream = s3SeekableInputStreamFactory.createStream(s3URI, OpenStreamInformation.DEFAULT);
     assertNotNull(inputStream);
   }
 
@@ -184,7 +183,7 @@ void testCreateStreamThrowsOnNullArgument() {
     assertThrows(
         NullPointerException.class,
         () -> {
-          s3SeekableInputStreamFactory.createStream(null, mock(OpenStreamInformation.class));
+          s3SeekableInputStreamFactory.createStream(null, OpenStreamInformation.DEFAULT);
         });
   }
 
@@ -215,19 +214,17 @@ void testCreateLogicalIO() throws IOException {
         .storeObjectMetadata(testURITXT, objectMetadata);
 
     assertTrue(
-        s3SeekableInputStreamFactory.createLogicalIO(
-                testURIParquet, mock(OpenStreamInformation.class))
+        s3SeekableInputStreamFactory.createLogicalIO(testURIParquet, OpenStreamInformation.DEFAULT)
             instanceof ParquetLogicalIOImpl);
     assertTrue(
-        s3SeekableInputStreamFactory.createLogicalIO(
-                testURIKEYPAR, mock(OpenStreamInformation.class))
+        s3SeekableInputStreamFactory.createLogicalIO(testURIKEYPAR, OpenStreamInformation.DEFAULT)
             instanceof ParquetLogicalIOImpl);
 
     assertTrue(
-        s3SeekableInputStreamFactory.createLogicalIO(testURIJAVA, mock(OpenStreamInformation.class))
+        s3SeekableInputStreamFactory.createLogicalIO(testURIJAVA, OpenStreamInformation.DEFAULT)
             instanceof DefaultLogicalIOImpl);
     assertTrue(
-        s3SeekableInputStreamFactory.createLogicalIO(testURITXT, mock(OpenStreamInformation.class))
+        s3SeekableInputStreamFactory.createLogicalIO(testURITXT, OpenStreamInformation.DEFAULT)
             instanceof SequentialLogicalIOImpl);
   }
 
@@ -307,7 +304,7 @@ private static void assertInputStreamReadExceptions(
         new S3SeekableInputStreamFactory(
             new S3SdkObjectClient(mockS3AsyncClient), S3SeekableInputStreamConfiguration.DEFAULT);
     S3SeekableInputStream inputStream =
-        factory.createStream(TEST_URI, mock(OpenStreamInformation.class));
+        factory.createStream(TEST_URI, OpenStreamInformation.DEFAULT);
     Exception thrownException = assertThrows(Exception.class, inputStream::read);
     assertInstanceOf(IOException.class, thrownException);
     Optional.ofNullable(thrownException.getCause())
@@ -323,8 +320,7 @@ private static void assertInputStreamHeadException(
             new S3SdkObjectClient(mockS3AsyncClient), S3SeekableInputStreamConfiguration.DEFAULT);
     Exception thrownException =
         assertThrows(
-            Exception.class,
-            () -> factory.createStream(TEST_URI, mock(OpenStreamInformation.class)));
+            Exception.class, () -> factory.createStream(TEST_URI, OpenStreamInformation.DEFAULT));
     assertInstanceOf(IOException.class, thrownException);
     Optional.ofNullable(thrownException.getCause())
         .ifPresent(

input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/S3SeekableInputStreamTest.java

@@ -364,7 +364,11 @@ void testMultiThreadUsage() throws IOException, InterruptedException {
 
     FakeObjectClient fakeObjectClient = new FakeObjectClient(sb.toString());
     MetadataStore metadataStore =
-        new MetadataStore(fakeObjectClient, TestTelemetry.DEFAULT, PhysicalIOConfiguration.DEFAULT);
+        new MetadataStore(
+            fakeObjectClient,
+            TestTelemetry.DEFAULT,
+            PhysicalIOConfiguration.DEFAULT,
+            mock(Metrics.class));
     BlobStore blobStore =
         new BlobStore(
             fakeObjectClient,
@@ -564,7 +568,11 @@ private S3SeekableInputStream getTestStreamWithContent(String content, S3URI s3U
       throws IOException {
     FakeObjectClient fakeObjectClient = new FakeObjectClient(content);
     MetadataStore metadataStore =
-        new MetadataStore(fakeObjectClient, TestTelemetry.DEFAULT, PhysicalIOConfiguration.DEFAULT);
+        new MetadataStore(
+            fakeObjectClient,
+            TestTelemetry.DEFAULT,
+            PhysicalIOConfiguration.DEFAULT,
+            mock(Metrics.class));
     BlobStore blobStore =
         new BlobStore(
             fakeObjectClient,

input-stream/src/test/java/software/amazon/s3/analyticsaccelerator/S3SeekableInputStreamTestBase.java

@@ -41,7 +41,11 @@ public class S3SeekableInputStreamTestBase {
   protected final PhysicalIOConfiguration physicalIOConfiguration = PhysicalIOConfiguration.DEFAULT;
   protected final FakeObjectClient fakeObjectClient = new FakeObjectClient(TEST_DATA);
   protected final MetadataStore metadataStore =
-      new MetadataStore(fakeObjectClient, TestTelemetry.DEFAULT, PhysicalIOConfiguration.DEFAULT);
+      new MetadataStore(
+          fakeObjectClient,
+          TestTelemetry.DEFAULT,
+          PhysicalIOConfiguration.DEFAULT,
+          mock(Metrics.class));
   protected final BlobStore blobStore =
       new BlobStore(
           fakeObjectClient, TestTelemetry.DEFAULT, physicalIOConfiguration, mock(Metrics.class));