HADOOP-19295. S3A: large uploads can timeout over slow links

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

@@ -398,6 +398,21 @@ private Constants() {
   public static final Duration DEFAULT_CONNECTION_ACQUISITION_TIMEOUT_DURATION =
       Duration.ofSeconds(60);
 
+  /**
+   * Timeout for uploading all of a small object or a single part
+   * of a larger one.
+   * {@value}.
+   * Default unit is milliseconds for consistency with other options.
+   */
+  public static final String PART_UPLOAD_TIMEOUT =
+      "fs.s3a.connection.part.upload.timeout";
+
+  /**
+   * Default part upload timeout: 15 minutes.
+   */
+  public static final Duration DEFAULT_PART_UPLOAD_TIMEOUT =
+      Duration.ofMinutes(15);
+
   /**
    * Should TCP Keepalive be enabled on the socket?
    * This adds some network IO, but finds failures faster.

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

@@ -1286,6 +1286,13 @@ protected RequestFactory createRequestFactory() {
           STORAGE_CLASS);
     }
 
+    // optional custom timeout for bulk uploads
+    Duration partUploadTimeout = ConfigurationHelper.getDuration(getConf(),
+        PART_UPLOAD_TIMEOUT,
+        DEFAULT_PART_UPLOAD_TIMEOUT,
+        TimeUnit.MILLISECONDS,
+        Duration.ZERO);
+
     return RequestFactoryImpl.builder()
         .withBucket(requireNonNull(bucket))
         .withCannedACL(getCannedACL())
@@ -1295,6 +1302,7 @@ protected RequestFactory createRequestFactory() {
         .withContentEncoding(contentEncoding)
         .withStorageClass(storageClass)
         .withMultipartUploadEnabled(isMultipartUploadEnabled)
+        .withPartUploadTimeout(partUploadTimeout)
         .build();
   }
 

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

@@ -26,6 +26,8 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.awscore.AwsRequest;
+import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
 import software.amazon.awssdk.core.client.config.SdkAdvancedClientOption;
 import software.amazon.awssdk.core.retry.RetryMode;
@@ -623,4 +625,24 @@ static ConnectionSettings createConnectionSettings(Configuration conf) {
         socketTimeout);
   }
 
+  /**
+   * Set a custom ApiCallTimeout for a single request.
+   * This allows for a longer timeout to be used in data upload
+   * requests than that for all other S3 interactions;
+   * This does not happen by default in the V2 SDK
+   * (see HADOOP-19295).
+   * <p>
+   * If the timeout is zero, the request is not patched.
+   * @param builder builder to patch.
+   * @param timeout timeout
+   */
+  public static void setRequestTimeout(AwsRequest.Builder builder, Duration timeout) {
+    if (!timeout.isZero()) {
+      builder.overrideConfiguration(
+          AwsRequestOverrideConfiguration.builder()
+              .apiCallTimeout(timeout)
+              .apiCallAttemptTimeout(timeout)
+              .build());
+    }
+  }
 }

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

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.s3a.impl;
 
+import java.time.Duration;
 import java.util.Base64;
 import java.util.HashMap;
 import java.util.List;
@@ -60,6 +61,7 @@
 
 import static org.apache.commons.lang3.StringUtils.isNotEmpty;
 import static org.apache.hadoop.fs.s3a.S3AEncryptionMethods.UNKNOWN_ALGORITHM;
+import static org.apache.hadoop.fs.s3a.impl.AWSClientConfig.setRequestTimeout;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.DEFAULT_UPLOAD_PART_COUNT_LIMIT;
 import static org.apache.hadoop.util.Preconditions.checkArgument;
 import static org.apache.hadoop.util.Preconditions.checkNotNull;
@@ -128,6 +130,12 @@ public class RequestFactoryImpl implements RequestFactory {
    */
   private final boolean isMultipartUploadEnabled;
 
+  /**
+   * Timeout for uploading objects/parts.
+   * This will be set on data put/post operations only.
+   */
+  private final Duration partUploadTimeout;
+
   /**
    * Constructor.
    * @param builder builder with all the configuration.
@@ -142,6 +150,7 @@ protected RequestFactoryImpl(
     this.contentEncoding = builder.contentEncoding;
     this.storageClass = builder.storageClass;
     this.isMultipartUploadEnabled = builder.isMultipartUploadEnabled;
+    this.partUploadTimeout = builder.partUploadTimeout;
   }
 
   /**
@@ -344,6 +353,11 @@ public PutObjectRequest.Builder newPutObjectRequestBuilder(String key,
       putObjectRequestBuilder.storageClass(storageClass);
     }
 
+    // Set the timeout for object uploads but not directory markers.
+    if (!isDirectoryMarker) {
+      setRequestTimeout(putObjectRequestBuilder, partUploadTimeout);
+    }
+
     return prepareRequest(putObjectRequestBuilder);
   }
 
@@ -595,6 +609,9 @@ public UploadPartRequest.Builder newUploadPartRequestBuilder(
         .partNumber(partNumber)
         .contentLength(size);
     uploadPartEncryptionParameters(builder);
+
+    // Set the request timeout for the part upload
+    setRequestTimeout(builder, partUploadTimeout);
     return prepareRequest(builder);
   }
 
@@ -702,6 +719,13 @@ public static final class RequestFactoryBuilder {
      */
     private boolean isMultipartUploadEnabled = true;
 
+    /**
+     * Timeout for uploading objects/parts.
+     * This will be set on data put/post operations only.
+     * A zero value means "no custom timeout"
+     */
+    private Duration partUploadTimeout = Duration.ZERO;
+
     private RequestFactoryBuilder() {
     }
 
@@ -799,6 +823,18 @@ public RequestFactoryBuilder withMultipartUploadEnabled(
       this.isMultipartUploadEnabled = value;
       return this;
     }
+
+    /**
+     * Timeout for uploading objects/parts.
+     * This will be set on data put/post operations only.
+     * A zero value means "no custom timeout"
+     * @param value new value
+     * @return the builder
+     */
+    public RequestFactoryBuilder withPartUploadTimeout(final Duration value) {
+      partUploadTimeout = value;
+      return this;
+    }
   }
 
   /**

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

@@ -26,6 +26,7 @@
 import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.nio.ByteBuffer;
+import java.time.LocalDateTime;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
@@ -224,6 +225,12 @@ public static abstract class BaseContentProvider<T extends InputStream>
      */
     private T currentStream;
 
+    /**
+     * When did this upload start?
+     * Use in error messages.
+     */
+    private final LocalDateTime startTime;
+
     /**
      * Constructor.
      * @param size size of the data. Must be non-negative.
@@ -241,6 +248,7 @@ protected BaseContentProvider(int size, @Nullable Supplier<Boolean> isOpen) {
       checkArgument(size >= 0, "size is negative: %s", size);
       this.size = size;
       this.isOpen = isOpen;
+      this.startTime = LocalDateTime.now();
     }
 
     /**
@@ -275,7 +283,10 @@ public final InputStream newStream() {
       checkOpen();
       streamCreationCount++;
       if (streamCreationCount > 1) {
-        LOG.info("Stream created more than once: {}", this);
+        LOG.info("Stream recreated: {}", this);
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Stream creation stack", new Exception("here"));
+        }
       }
       return setCurrentStream(createNewStream());
     }
@@ -302,6 +313,14 @@ public int getSize() {
       return size;
     }
 
+    /**
+     * When did this upload start?
+     * @return start time
+     */
+    public LocalDateTime getStartTime() {
+      return startTime;
+    }
+
     /**
      * Current stream.
      * When {@link #newStream()} is called, this is set to the new value,
@@ -330,6 +349,7 @@ protected T setCurrentStream(T stream) {
     public String toString() {
       return "BaseContentProvider{" +
           "size=" + size +
+          ", initiated at " + startTime +
           ", streamCreationCount=" + streamCreationCount +
           ", currentStream=" + currentStream +
           '}';

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestUploadRecovery.java

@@ -153,15 +153,15 @@ public Configuration createConfiguration() {
    */
   @Override
   public void setup() throws Exception {
-    SdkFaultInjector.resetEvaluator();
+    SdkFaultInjector.resetFaultInjector();
     super.setup();
   }
 
   @Override
   public void teardown() throws Exception {
     // safety check in case the evaluation is failing any
     // request needed in cleanup.
-    SdkFaultInjector.resetEvaluator();
+    SdkFaultInjector.resetFaultInjector();
 
     super.teardown();
   }