don't add request level timeout if CRT is enabled

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

@@ -474,6 +474,11 @@ public class S3AFileSystem extends FileSystem implements StreamCapabilities,
    */
   private boolean fipsEnabled;
 
+  /**
+   * Is CRT enabled?
+   */
+  private boolean isCRTEnabled;
+
   /**
    * A cache of files that should be deleted when the FileSystem is closed
    * or the JVM is exited.
@@ -636,6 +641,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());
 
+      isCRTEnabled = conf.getBoolean(CRT_CLIENT_ENABLED, DEFAULT_CRT_ENABLED);
+
       isAnalyticsAcceleratorEnabled = StreamIntegration.determineInputStreamType(conf)
           .equals(InputStreamType.Analytics);
 
@@ -1168,7 +1175,7 @@ private ClientManager createClientManager(URI fsURI, boolean dtEnabled) throws I
         .withClientSideEncryptionMaterials(cseMaterials)
         .withAnalyticsAcceleratorEnabled(isAnalyticsAcceleratorEnabled)
         .withKMSRegion(conf.get(S3_ENCRYPTION_CSE_KMS_REGION))
-        .withCrtEnabled(conf.getBoolean(CRT_CLIENT_ENABLED, DEFAULT_CRT_ENABLED));
+        .withCrtEnabled(isCRTEnabled);
 
     // this is where clients and the transfer manager are created on demand.
     return createClientManager(clientFactory, unencryptedClientFactory, parameters,
@@ -1328,6 +1335,7 @@ protected RequestFactory createRequestFactory() {
         .withStorageClass(storageClass)
         .withMultipartUploadEnabled(isMultipartUploadEnabled)
         .withPartUploadTimeout(partUploadTimeout)
+        .withCRTEnabled(isCRTEnabled)
         .build();
   }
 

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

@@ -250,6 +250,12 @@ public static RetryPolicy.Builder createRetryPolicyBuilder(Configuration conf) {
 
   private static S3CrtProxyConfiguration mapCRTProxyConfiguration(software.amazon.awssdk.http.nio.netty.ProxyConfiguration
       proxyConfiguration) {
+
+    // If no proxy configurations are set, then proxy configuration can be null.
+    if (proxyConfiguration == null) {
+      return null;
+    }
+
     S3CrtProxyConfiguration.Builder builder = S3CrtProxyConfiguration.builder();
 
     builder.host(proxyConfiguration.host());

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

@@ -138,6 +138,11 @@ public class RequestFactoryImpl implements RequestFactory {
    */
   private final Duration partUploadTimeout;
 
+  /**
+   * Is the S3 CRT client enabled?
+   */
+  private final Boolean isCRTEnabled;
+
   /**
    * Constructor.
    * @param builder builder with all the configuration.
@@ -153,6 +158,7 @@ protected RequestFactoryImpl(
     this.storageClass = builder.storageClass;
     this.isMultipartUploadEnabled = builder.isMultipartUploadEnabled;
     this.partUploadTimeout = builder.partUploadTimeout;
+    this.isCRTEnabled = builder.isCRTEnabled;
   }
 
   /**
@@ -356,7 +362,7 @@ public PutObjectRequest.Builder newPutObjectRequestBuilder(String key,
     }
 
     // Set the timeout for object uploads but not directory markers.
-    if (!isDirectoryMarker) {
+    if (!isDirectoryMarker && !isCRTEnabled) {
       setRequestTimeout(putObjectRequestBuilder, partUploadTimeout);
     }
 
@@ -725,6 +731,11 @@ public static final class RequestFactoryBuilder {
      */
     private boolean isMultipartUploadEnabled = true;
 
+    /**
+     * Is the S3 CRT client enabled?
+     */
+    private boolean isCRTEnabled;
+
     /**
      * Timeout for uploading objects/parts.
      * This will be set on data put/post operations only.
@@ -841,6 +852,11 @@ public RequestFactoryBuilder withPartUploadTimeout(final Duration value) {
       partUploadTimeout = value;
       return this;
     }
+
+    public RequestFactoryBuilder withCRTEnabled(final boolean value) {
+      isCRTEnabled = value;
+      return this;
+    }
   }
 
   /**