Add retry support for removeObjects()/completeMultipartUpload() APIs (#1704)

Signed-off-by: Bala.FA <bala@minio.io>

Author: balamurugana

adminapi/src/main/java/io/minio/admin/MinioAdminClient.java

@@ -46,19 +46,20 @@
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
-import java.util.Locale;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
+import java.util.stream.IntStream;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
+import okhttp3.Interceptor;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.RequestBody;
 import okhttp3.Response;
-import okhttp3.ResponseBody;
 
 /** Client to perform MinIO administration operations. */
 public class MinioAdminClient {
@@ -111,12 +112,12 @@ public String toString() {
   }
 
   private String userAgent = Utils.getDefaultUserAgent();
-  private PrintWriter traceStream;
+  private volatile PrintWriter traceStream;
 
   private HttpUrl baseUrl;
   private String region;
   private Provider provider;
-  private OkHttpClient httpClient;
+  private volatile OkHttpClient httpClient;
 
   private MinioAdminClient(
       HttpUrl baseUrl, String region, Provider provider, OkHttpClient httpClient) {
@@ -132,6 +133,39 @@ private Credentials getCredentials() throws MinioException {
     return creds;
   }
 
+  private static int getStatusRetryInterceptorIndex(List<Interceptor> interceptors) {
+    return IntStream.range(0, interceptors.size())
+        .filter(i -> interceptors.get(i) instanceof Http.StatusRetryInterceptor)
+        .findFirst()
+        .orElse(-1);
+  }
+
+  private OkHttpClient getHttpClient(PrintWriter traceStream) {
+    if (traceStream == null) return this.httpClient;
+
+    OkHttpClient httpClient = this.httpClient;
+    List<Interceptor> interceptors = httpClient.interceptors();
+    int i = getStatusRetryInterceptorIndex(interceptors);
+    Http.StatusRetryInterceptor interceptor =
+        i < 0 ? null : (Http.StatusRetryInterceptor) interceptors.get(i);
+
+    OkHttpClient.Builder builder = httpClient.newBuilder();
+    if (interceptor == null) {
+      builder.addInterceptor(new Http.StatusRetryInterceptor(interceptor, traceStream, false));
+    } else {
+      builder.interceptors().clear();
+      for (int j = 0; j < interceptors.size(); j++) {
+        if (i == j) {
+          builder.addInterceptor(new Http.StatusRetryInterceptor(interceptor, traceStream, false));
+        } else {
+          builder.addInterceptor(interceptors.get(j));
+        }
+      }
+    }
+
+    return builder.build();
+  }
+
   private Response httpExecute(
       Http.Method method, Command command, Multimap<String, String> queryParamMap, byte[] body)
       throws IOException, MinioException {
@@ -179,40 +213,8 @@ private Response httpExecute(
             creds.secretKey(),
             request.header("x-amz-content-sha256"));
 
-    PrintWriter traceStream = this.traceStream;
-    if (traceStream != null) {
-      StringBuilder traceBuilder = new StringBuilder();
-      traceBuilder.append("---------START-HTTP---------\n");
-      String encodedPath = request.url().encodedPath();
-      String encodedQuery = request.url().encodedQuery();
-      if (encodedQuery != null) encodedPath += "?" + encodedQuery;
-      traceBuilder.append(request.method()).append(" ").append(encodedPath).append(" HTTP/1.1\n");
-      traceBuilder.append(
-          request
-              .headers()
-              .toString()
-              .replaceAll("Signature=([0-9a-f]+)", "Signature=*REDACTED*")
-              .replaceAll("Credential=([^/]+)", "Credential=*REDACTED*"));
-      if (body != null) traceBuilder.append("\n").append(new String(body, StandardCharsets.UTF_8));
-      traceStream.println(traceBuilder.toString());
-    }
-
-    OkHttpClient httpClient = this.httpClient;
+    OkHttpClient httpClient = getHttpClient(this.traceStream);
     Response response = httpClient.newCall(request).execute();
-
-    if (traceStream != null) {
-      String trace =
-          response.protocol().toString().toUpperCase(Locale.US)
-              + " "
-              + response.code()
-              + "\n"
-              + response.headers();
-      traceStream.println(trace);
-      ResponseBody responseBody = response.peekBody(1024 * 1024);
-      traceStream.println(responseBody.string());
-      traceStream.println("----------END-HTTP----------");
-    }
-
     if (response.isSuccessful()) return response;
 
     throw new MinioException("Request failed with response: " + response.body().string());
@@ -926,6 +928,38 @@ public void traceOff() {
     this.traceStream = null;
   }
 
+  /**
+   * Sets request retry parameters. Any null/invalid values disable retry.
+   *
+   * <pre>Example:{@code
+   * minioClient.setRetry(ImmutableSet.of(408, 504), 250, 3);
+   * }</pre>
+   *
+   * @param retryStatusCodes HTTP status codes to be retried.
+   * @param delayMs Delay between retries.
+   * @param maxRetries Maximum number of retry attempts.
+   */
+  public synchronized void setRetry(
+      Set<Integer> retryStatusCodes, Long delayMs, Integer maxRetries) {
+    Interceptor interceptor =
+        new Http.StatusRetryInterceptor(
+            retryStatusCodes, delayMs == null ? 0 : delayMs, maxRetries == null ? 0 : maxRetries);
+
+    List<Interceptor> interceptors = this.httpClient.interceptors();
+    int i = getStatusRetryInterceptorIndex(interceptors);
+    OkHttpClient.Builder builder = this.httpClient.newBuilder();
+    if (i >= 0) {
+      builder.interceptors().clear();
+      for (int j = 0; j < interceptors.size(); j++) {
+        builder.addInterceptor(i == j ? interceptor : interceptors.get(j));
+      }
+    } else {
+      builder.addInterceptor(interceptor);
+    }
+
+    this.httpClient = builder.build();
+  }
+
   public static Builder builder() {
     return new Builder();
   }

api/src/main/java/io/minio/BaseS3Client.java

@@ -21,6 +21,7 @@
 import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.json.JsonMapper;
+import com.google.common.collect.ImmutableSet;
 import io.minio.credentials.Credentials;
 import io.minio.credentials.Provider;
 import io.minio.errors.ErrorResponseException;
@@ -60,6 +61,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Supplier;
 import java.util.logging.Logger;
 import java.util.stream.IntStream;
@@ -84,6 +86,8 @@ public abstract class BaseS3Client implements AutoCloseable {
     }
   }
 
+  protected static final Set<String> RETRYABLE_ERRORS =
+      ImmutableSet.of("InternalError", "RequestTimeout", "ServiceUnavailable", "SlowDown");
   protected static final String NO_SUCH_BUCKET_MESSAGE = "Bucket does not exist";
   protected static final String NO_SUCH_BUCKET = "NoSuchBucket";
   protected static final String NO_SUCH_BUCKET_POLICY = "NoSuchBucketPolicy";
@@ -675,24 +679,8 @@ public CompletableFuture<AbortMultipartUploadResponse> abortMultipartUpload(
             });
   }
 
-  /**
-   * Do <a
-   * href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html">CompleteMultipartUpload
-   * S3 API</a> asynchronously.
-   *
-   * @param args {@link CompleteMultipartUploadArgs} object.
-   * @return {@link CompletableFuture}&lt;{@link ObjectWriteResponse}&gt; object.
-   */
-  public CompletableFuture<ObjectWriteResponse> completeMultipartUpload(
-      CompleteMultipartUploadArgs args) {
-    checkArgs(args);
-    args.validateSsec(baseUrl.isHttps());
-    Http.Body body = null;
-    try {
-      body = new Http.Body(new CompleteMultipartUpload(args.parts()), null, null, null);
-    } catch (MinioException e) {
-      return Utils.failedFuture(e);
-    }
+  private CompletableFuture<ObjectWriteResponse> completeMultipartUpload(
+      CompleteMultipartUploadArgs args, Http.Body body) {
     return executePostAsync(
             args,
             args.ssec() == null ? null : args.ssec().headers(),
@@ -750,6 +738,59 @@ public CompletableFuture<ObjectWriteResponse> completeMultipartUpload(
             });
   }
 
+  private CompletableFuture<ObjectWriteResponse> completeMultipartUpload(
+      CompleteMultipartUploadArgs args, Http.Body body, int attempt) {
+    return completeMultipartUpload(args, body)
+        .handle(
+            (response, ex) -> {
+              if (ex == null) return CompletableFuture.completedFuture(response);
+
+              ex = ex.getCause();
+              if (attempt == Math.max(1, args.maxRetries()) - 1
+                  || !(ex instanceof ErrorResponseException)
+                  || !RETRYABLE_ERRORS.contains(
+                      ((ErrorResponseException) ex).errorResponse().code())) {
+                return Utils.<ObjectWriteResponse>failedFuture(ex);
+              }
+
+              if (args.delayMs() > 0) {
+                long maxBackoffLimit = args.delayMs() * (1L << (attempt + 1));
+                long jitteredDelay = ThreadLocalRandom.current().nextLong(0, maxBackoffLimit);
+                try {
+                  Thread.sleep(jitteredDelay);
+                } catch (InterruptedException e) {
+                  Thread.currentThread().interrupt();
+                  return Utils.<ObjectWriteResponse>failedFuture(
+                      new IllegalStateException("Retry timeout interrupted", e));
+                }
+              }
+
+              return completeMultipartUpload(args, body, attempt + 1);
+            })
+        .thenCompose(future -> future);
+  }
+
+  /**
+   * Do <a
+   * href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html">CompleteMultipartUpload
+   * S3 API</a> asynchronously.
+   *
+   * @param args {@link CompleteMultipartUploadArgs} object.
+   * @return {@link CompletableFuture}&lt;{@link ObjectWriteResponse}&gt; object.
+   */
+  public CompletableFuture<ObjectWriteResponse> completeMultipartUpload(
+      CompleteMultipartUploadArgs args) {
+    checkArgs(args);
+    args.validateSsec(baseUrl.isHttps());
+    Http.Body body = null;
+    try {
+      body = new Http.Body(new CompleteMultipartUpload(args.parts()), null, null, null);
+    } catch (MinioException e) {
+      return Utils.failedFuture(e);
+    }
+    return completeMultipartUpload(args, body, 0);
+  }
+
   /**
    * Do <a href="https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html">CopyObject S3
    * API</a> asynchronously.

api/src/main/java/io/minio/CompleteMultipartUploadArgs.java

@@ -25,13 +25,17 @@ public class CompleteMultipartUploadArgs extends ObjectArgs {
   private String uploadId;
   private Part[] parts;
   private ServerSideEncryption.CustomerKey ssec;
+  private long delayMs = 100L;
+  private int maxRetries = 5;
 
   protected CompleteMultipartUploadArgs() {}
 
   public CompleteMultipartUploadArgs(ComposeObjectArgs args, String uploadId, Part[] parts) {
     super(args);
     this.uploadId = uploadId;
     this.parts = parts;
+    this.delayMs = args.delayMs();
+    this.maxRetries = args.maxRetries();
   }
 
   public CompleteMultipartUploadArgs(PutObjectBaseArgs args, String uploadId, Part[] parts) {
@@ -41,6 +45,8 @@ public CompleteMultipartUploadArgs(PutObjectBaseArgs args, String uploadId, Part
     if (args.sse() != null && args.sse() instanceof ServerSideEncryption.CustomerKey) {
       this.ssec = (ServerSideEncryption.CustomerKey) args.sse();
     }
+    this.delayMs = args.delayMs();
+    this.maxRetries = args.maxRetries();
   }
 
   public String uploadId() {
@@ -55,6 +61,14 @@ public ServerSideEncryption.CustomerKey ssec() {
     return ssec;
   }
 
+  public long delayMs() {
+    return delayMs;
+  }
+
+  public int maxRetries() {
+    return maxRetries;
+  }
+
   public void validateSsec(boolean isHttps) {
     checkSse(ssec, isHttps);
   }
@@ -89,6 +103,18 @@ public Builder ssec(ServerSideEncryption.CustomerKey ssec) {
       operations.add(args -> args.ssec = ssec);
       return this;
     }
+
+    /** Set delay between retries. Value <= 0 makes no delay (default 100ms). */
+    public Builder delayMs(long delayMs) {
+      operations.add(args -> args.delayMs = delayMs);
+      return this;
+    }
+
+    /** Set maximum retry between failure. Value <= 0 disables retry (default 5). */
+    public Builder maxRetries(int maxRetries) {
+      operations.add(args -> args.maxRetries = maxRetries);
+      return this;
+    }
   }
 
   @Override
@@ -99,11 +125,13 @@ public boolean equals(Object o) {
     CompleteMultipartUploadArgs that = (CompleteMultipartUploadArgs) o;
     return Objects.equals(uploadId, that.uploadId)
         && Arrays.equals(parts, that.parts)
-        && Objects.equals(ssec, that.ssec);
+        && Objects.equals(ssec, that.ssec)
+        && delayMs == that.delayMs
+        && maxRetries == that.maxRetries;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(super.hashCode(), uploadId, parts, ssec);
+    return Objects.hash(super.hashCode(), uploadId, parts, ssec, delayMs, maxRetries);
   }
 }