only enable multipart on async client if AAL && CSE are disabled.

This commit should not be part of the final merge, but is required till we upgrade the SDK.

Author: ahmarsuhail

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

@@ -166,7 +166,7 @@ public S3AsyncClient createS3AsyncClient(
                 .httpClientBuilder(httpClientBuilder);
 
     // multipart upload pending with HADOOP-19326.
-    if (!parameters.isClientSideEncryptionEnabled()) {
+    if (!parameters.isClientSideEncryptionEnabled() && !parameters.isAnalyticsAcceleratorEnabled()) {
       s3AsyncClientBuilder.multipartConfiguration(multipartConfiguration)
               .multipartEnabled(parameters.isMultipartCopy());
     }

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

@@ -146,9 +146,11 @@
 import org.apache.hadoop.fs.s3a.impl.StoreContextFactory;
 import org.apache.hadoop.fs.s3a.impl.UploadContentProviders;
 import org.apache.hadoop.fs.s3a.impl.CSEUtils;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectReadParameters;
 import org.apache.hadoop.fs.s3a.impl.streams.ObjectInputStreamCallbacks;
 import org.apache.hadoop.fs.s3a.impl.streams.StreamFactoryRequirements;
+import org.apache.hadoop.fs.s3a.impl.streams.StreamIntegration;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperations;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperationsImpl;
 import org.apache.hadoop.fs.statistics.DurationTracker;
@@ -433,6 +435,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
    */
   private boolean isCSEEnabled;
 
+  /**
+   * Is this S3A FS instance using analytics accelerator?
+   */
+  private boolean isAnalyticsAccelaratorEnabled;
+
   /**
    * Bucket AccessPoint.
    */
@@ -605,6 +612,8 @@ public void initialize(URI name, Configuration originalConf)
       // If encryption method is set to CSE-KMS or CSE-CUSTOM then CSE is enabled.
       isCSEEnabled = CSEUtils.isCSEEnabled(getS3EncryptionAlgorithm().getMethod());
 
+      isAnalyticsAccelaratorEnabled = StreamIntegration.determineInputStreamType(conf).equals(InputStreamType.Analytics);
+
       // Create the appropriate fsHandler instance using a factory method
       fsHandler = createFileSystemHandler();
       fsHandler.setCSEGauge((IOStatisticsStore) getIOStatistics());
@@ -1121,6 +1130,7 @@ private ClientManager createClientManager(URI fsURI, boolean dtEnabled) throws I
             conf.getBoolean(CHECKSUM_VALIDATION, CHECKSUM_VALIDATION_DEFAULT))
         .withClientSideEncryptionEnabled(isCSEEnabled)
         .withClientSideEncryptionMaterials(cseMaterials)
+        .withAnalyticsAcceleratorEnabled(isAnalyticsAccelaratorEnabled)
         .withKMSRegion(conf.get(S3_ENCRYPTION_CSE_KMS_REGION));
 
     // this is where clients and the transfer manager are created on demand.

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

@@ -202,6 +202,11 @@ final class S3ClientCreationParameters {
      */
     private boolean fipsEnabled;
 
+    /**
+     * Is analytics accelerator enabled
+     */
+    private boolean isAnalyticsAcceleratorEnabled;
+
     /**
      * List of execution interceptors to include in the chain
      * of interceptors in the SDK.
@@ -457,6 +462,17 @@ public S3ClientCreationParameters withClientSideEncryptionEnabled(final boolean
       return this;
     }
 
+    /**
+     * Set the analytics accelerator enabled flag.
+     *
+     * @param value new value
+     * @return the builder
+     */
+    public S3ClientCreationParameters withAnalyticsAcceleratorEnabled(final boolean value) {
+      this.isAnalyticsAcceleratorEnabled = value;
+      return this;
+    }
+
     /**
      * Set the KMS client region.
      * This is required for CSE-KMS
@@ -477,6 +493,14 @@ public boolean isClientSideEncryptionEnabled() {
       return this.isCSEEnabled;
     }
 
+    /**
+     * Get the analytics accelerator enabled flag.
+     * @return analytics accelerator enabled flag.
+     */
+    public boolean isAnalyticsAcceleratorEnabled() {
+      return this.isAnalyticsAcceleratorEnabled;
+    }
+
     /**
      * Set the client side encryption materials.
      *

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

@@ -135,7 +135,7 @@ public static ObjectInputStreamFactory factoryFromConfig(final Configuration con
    * @param conf configuration
    * @return a stream factory.
    */
-  static InputStreamType determineInputStreamType(final Configuration conf) {
+  public static InputStreamType determineInputStreamType(final Configuration conf) {
     // work out the default stream; this includes looking at the
     // deprecated prefetch enabled key to see if it is set.
     if (conf.getBoolean(PREFETCH_ENABLED_KEY, false)) {