HADOOP-19664. S3A: Analytics stream to use Java sync client. (#7909)

Contributed by: Ahmar Suhail

Author: ahmarsuhail

hadoop-project/pom.xml

@@ -219,7 +219,7 @@
     <aws-java-sdk.version>1.12.720</aws-java-sdk.version>
     <aws-java-sdk-v2.version>2.29.52</aws-java-sdk-v2.version>
     <amazon-s3-encryption-client-java.version>3.1.1</amazon-s3-encryption-client-java.version>
-    <amazon-s3-analyticsaccelerator-s3.version>1.2.1</amazon-s3-analyticsaccelerator-s3.version>
+    <amazon-s3-analyticsaccelerator-s3.version>1.3.0</amazon-s3-analyticsaccelerator-s3.version>
     <aws.eventstream.version>1.0.1</aws.eventstream.version>
     <hsqldb.version>2.7.1</hsqldb.version>
     <frontend-maven-plugin.version>1.11.2</frontend-maven-plugin.version>

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/S3AStoreImpl.java

@@ -993,10 +993,10 @@ public InputStreamType streamType() {
   private class FactoryCallbacks implements StreamFactoryCallbacks {
 
     @Override
-    public S3AsyncClient getOrCreateAsyncClient(final boolean requireCRT) throws IOException {
+    public S3Client getOrCreateSyncClient() throws IOException {
       // Needs support of the CRT before the requireCRT can be used
       LOG.debug("Stream factory requested async client");
-      return clientManager().getOrCreateAsyncClient();
+      return clientManager().getOrCreateS3Client();
     }
 
     @Override

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsStreamFactory.java

@@ -23,9 +23,10 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import software.amazon.s3.analyticsaccelerator.S3SdkObjectClient;
+
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamConfiguration;
 import software.amazon.s3.analyticsaccelerator.S3SeekableInputStreamFactory;
+import software.amazon.s3.analyticsaccelerator.S3SyncSdkObjectClient;
 import software.amazon.s3.analyticsaccelerator.common.ConnectorConfiguration;
 
 import org.apache.hadoop.conf.Configuration;
@@ -118,7 +119,7 @@ private S3SeekableInputStreamFactory getOrCreateS3SeekableInputStreamFactory()
 
   private CallableRaisingIOE<S3SeekableInputStreamFactory> createS3SeekableInputStreamFactory() {
     return () -> new S3SeekableInputStreamFactory(
-            new S3SdkObjectClient(callbacks().getOrCreateAsyncClient(requireCrt)),
+            new S3SyncSdkObjectClient(callbacks().getOrCreateSyncClient()),
             seekableInputStreamConfiguration);
   }
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/ObjectInputStreamFactory.java

@@ -20,7 +20,7 @@
 
 import java.io.IOException;
 
-import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
 
 import org.apache.hadoop.fs.s3a.Statistic;
 import org.apache.hadoop.fs.StreamCapabilities;
@@ -80,12 +80,11 @@ ObjectInputStream readObject(ObjectReadParameters parameters)
   interface StreamFactoryCallbacks {
 
     /**
-     * Get the Async S3Client, raising a failure to create as an IOException.
-     * @param requireCRT is the CRT required.
+     * Get the Sync S3Client, raising a failure to create as an IOException.
      * @return the Async S3 client
      * @throws IOException failure to create the client.
      */
-    S3AsyncClient getOrCreateAsyncClient(boolean requireCRT) throws IOException;
+    S3Client getOrCreateSyncClient() throws IOException;
 
     void incrementFactoryStatistic(Statistic statistic);
   }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/streams/TestStreamFactories.java

@@ -23,12 +23,12 @@
 
 import org.apache.hadoop.fs.s3a.Statistic;
 import org.junit.jupiter.api.Test;
-import software.amazon.awssdk.services.s3.S3AsyncClient;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.s3a.VectoredIOContext;
 import org.apache.hadoop.fs.s3a.prefetch.PrefetchingInputStreamFactory;
 import org.apache.hadoop.test.AbstractHadoopTestBase;
+import software.amazon.awssdk.services.s3.S3Client;
 
 import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_CUSTOM_FACTORY;
 import static org.apache.hadoop.fs.s3a.Constants.INPUT_STREAM_TYPE;
@@ -329,9 +329,8 @@ public FactoryFailsToInstantiate() {
    * Callbacks from {@link ObjectInputStreamFactory} instances.
    */
   private static final class Callbacks implements ObjectInputStreamFactory.StreamFactoryCallbacks {
-
     @Override
-    public S3AsyncClient getOrCreateAsyncClient(final boolean requireCRT) throws IOException {
+    public S3Client getOrCreateSyncClient() throws IOException {
       throw new UnsupportedOperationException("not implemented");
     }