HADOOP-19992. Retry OSS multipart completion on QpsLimitExceeded.

hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/AliyunOSSFileSystemStore.java

@@ -55,6 +55,10 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.io.retry.RetryPolicies;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.io.retry.RetryPolicy.RetryAction;
+import org.apache.hadoop.util.Preconditions;
 import org.apache.hadoop.util.VersionInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -72,6 +76,7 @@
 import java.util.List;
 import java.util.ListIterator;
 import java.util.NoSuchElementException;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import com.aliyun.oss.common.comm.SignVersion;
 
@@ -85,6 +90,8 @@
 public class AliyunOSSFileSystemStore {
   public static final Logger LOG =
       LoggerFactory.getLogger(AliyunOSSFileSystemStore.class);
+  private static final int MAX_MULTIPART_COMPLETE_RETRY_LIMIT = 10;
+  private static final long MAX_MULTIPART_COMPLETE_RETRY_DELAY = 10_000;
   private String username;
   private FileSystem.Statistics statistics;
   private OSSClient ossClient;
@@ -93,6 +100,7 @@ public class AliyunOSSFileSystemStore {
   private int maxKeys;
   private String serverSideEncryptionAlgorithm;
   private boolean useListV1;
+  private RetryPolicy multipartCompleteRetryPolicy;
 
   public void initialize(URI uri, Configuration conf, String user,
                          FileSystem.Statistics stat) throws IOException {
@@ -106,6 +114,21 @@ public void initialize(URI uri, Configuration conf, String user,
     clientConf.setProtocol(secureConnections ? Protocol.HTTPS : Protocol.HTTP);
     clientConf.setMaxErrorRetry(conf.getInt(MAX_ERROR_RETRIES_KEY,
         MAX_ERROR_RETRIES_DEFAULT));
+    int retryLimit = conf.getInt(MULTIPART_COMPLETE_RETRY_LIMIT_KEY,
+        MULTIPART_COMPLETE_RETRY_LIMIT_DEFAULT);
+    Preconditions.checkArgument(retryLimit >= 0
+            && retryLimit <= MAX_MULTIPART_COMPLETE_RETRY_LIMIT,
+        "%s must be between 0 and %s", MULTIPART_COMPLETE_RETRY_LIMIT_KEY,
+        MAX_MULTIPART_COMPLETE_RETRY_LIMIT);
+    long retryInterval = conf.getTimeDuration(
+        MULTIPART_COMPLETE_RETRY_INTERVAL_KEY,
+        MULTIPART_COMPLETE_RETRY_INTERVAL_DEFAULT, TimeUnit.MILLISECONDS);
+    Preconditions.checkArgument(retryInterval > 0
+            && retryInterval <= (Long.MAX_VALUE >> retryLimit),
+        "%s must be positive and not overflow after %s retries",
+        MULTIPART_COMPLETE_RETRY_INTERVAL_KEY, retryLimit);
+    multipartCompleteRetryPolicy = RetryPolicies.exponentialBackoffRetry(
+        retryLimit, retryInterval, TimeUnit.MILLISECONDS);
     clientConf.setConnectionTimeout(conf.getInt(ESTABLISH_TIMEOUT_KEY,
         ESTABLISH_TIMEOUT_DEFAULT));
     clientConf.setSocketTimeout(conf.getInt(SOCKET_TIMEOUT_KEY,
@@ -399,11 +422,8 @@ private boolean multipartCopy(String srcKey, long contentLength,
         statistics.incrementBytesWritten(size);
         partETags.add(partCopyResult.getPartETag());
       }
-      CompleteMultipartUploadRequest completeMultipartUploadRequest =
-          new CompleteMultipartUploadRequest(bucketName, dstKey,
-              uploadId, partETags);
       CompleteMultipartUploadResult completeMultipartUploadResult =
-          ossClient.completeMultipartUpload(completeMultipartUploadRequest);
+          completeMultipartUpload(dstKey, uploadId, partETags);
       LOG.debug(completeMultipartUploadResult.getETag());
       return true;
     } catch (OSSException | ClientException e) {
@@ -763,7 +783,47 @@ public CompleteMultipartUploadResult completeMultipartUpload(String key,
     CompleteMultipartUploadRequest completeMultipartUploadRequest =
         new CompleteMultipartUploadRequest(bucketName, key, uploadId,
             partETags);
-    return ossClient.completeMultipartUpload(completeMultipartUploadRequest);
+    for (int retries = 0; ; retries++) {
+      try {
+        return submitCompleteMultipartUpload(completeMultipartUploadRequest);
+      } catch (OSSException e) {
+        // A response with this error code means OSS rejected the request.
+        // Do not retry timeouts or other failures: Complete is not idempotent.
+        if (!"QpsLimitExceeded".equals(e.getErrorCode())) {
+          throw e;
+        }
+        RetryAction action;
+        try {
+          action = multipartCompleteRetryPolicy.shouldRetry(
+              e, retries, 0, false);
+        } catch (Exception policyFailure) {
+          e.addSuppressed(policyFailure);
+          throw e;
+        }
+        if (action.action != RetryAction.RetryDecision.RETRY) {
+          throw e;
+        }
+        LOG.warn("OSS throttled multipart completion of {} (retry {})",
+            key, retries + 1);
+        try {
+          sleepBeforeCompleteRetry(Math.min(action.delayMillis,
+              MAX_MULTIPART_COMPLETE_RETRY_DELAY));
+        } catch (InterruptedException interrupted) {
+          Thread.currentThread().interrupt();
+          throw new ClientException("Interrupted while retrying OSS multipart "
+              + "completion of " + key, interrupted);
+        }
+      }
+    }
+  }
+
+  CompleteMultipartUploadResult submitCompleteMultipartUpload(
+      CompleteMultipartUploadRequest request) {
+    return ossClient.completeMultipartUpload(request);
+  }
+
+  void sleepBeforeCompleteRetry(long delayMillis) throws InterruptedException {
+    Thread.sleep(delayMillis);
   }
 
   /**

hadoop-tools/hadoop-aliyun/src/main/java/org/apache/hadoop/fs/aliyun/oss/Constants.java

@@ -70,6 +70,14 @@ private Constants() {
   public static final String MAX_ERROR_RETRIES_KEY = "fs.oss.attempts.maximum";
   public static final int MAX_ERROR_RETRIES_DEFAULT = 10;
 
+  // Retry only a CompleteMultipartUpload rejected by OSS throttling.
+  public static final String MULTIPART_COMPLETE_RETRY_LIMIT_KEY =
+      "fs.oss.multipart.complete.retry.limit";
+  public static final int MULTIPART_COMPLETE_RETRY_LIMIT_DEFAULT = 3;
+  public static final String MULTIPART_COMPLETE_RETRY_INTERVAL_KEY =
+      "fs.oss.multipart.complete.retry.interval";
+  public static final long MULTIPART_COMPLETE_RETRY_INTERVAL_DEFAULT = 500;
+
   // Time until we give up trying to establish a connection to oss
   public static final String ESTABLISH_TIMEOUT_KEY =
       "fs.oss.connection.establish.timeout";

hadoop-tools/hadoop-aliyun/src/site/markdown/tools/hadoop-aliyun/index.md

@@ -168,6 +168,27 @@ please raise your issues with them.
       <description>How many times we should retry commands on transient errors.</description>
     </property>
 
+    <property>
+      <name>fs.oss.multipart.complete.retry.limit</name>
+      <value>3</value>
+      <description>Extra attempts after OSS explicitly rejects a multipart completion
+        with QpsLimitExceeded, for both uploads and copies. Must be between 0 and 10;
+        0 disables these retries. The initial request is not counted, so the default
+        allows at most 4 completion requests. This does not change SDK retries
+        configured by fs.oss.attempts.maximum for other operations.</description>
+    </property>
+
+    <property>
+      <name>fs.oss.multipart.complete.retry.interval</name>
+      <value>500ms</value>
+      <description>Base interval for exponential backoff with jitter when
+        retrying a throttled multipart completion. Must be at least 1ms.
+        Each retry delay, including jitter, is capped at 10s.
+        With the default settings, the three retry delays total at most 10.5s,
+        excluding request execution time.
+        Other completion failures are not retried.</description>
+    </property>
+
     <property>
       <name>fs.oss.connection.establish.timeout</name>
       <value>50000</value>