fix(s3wal): 503 SlowDown during WAL recovery from S3

s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java

@@ -935,6 +935,34 @@ protected <T> boolean bucketCheck(int bucketId, CompletableFuture<T> cf) {
         return true;
     }
 
+    protected <T> CompletableFuture<T> retryOnThrottle(Supplier<CompletableFuture<T>> operation,
+        S3Operation operationType, int retryCount) {
+
+        CompletableFuture<T> cf = new CompletableFuture<>();
+        operation.get().whenComplete((result, ex) -> {
+            if (ex == null) {
+                cf.complete(result);
+                return;
+            }
+
+            Throwable cause = FutureUtil.cause(ex);
+            if (!isThrottled(cause, retryCount)) {
+                cf.completeExceptionally(cause);
+                return;
+            }
+
+            int delay = retryDelay(operationType, retryCount);
+            scheduler.schedule(() -> 
+                retryOnThrottle(operation, operationType, retryCount + 1)
+                    .whenComplete((r, e) -> {
+                        if (e != null) cf.completeExceptionally(e);
+                        else cf.complete(r);
+                    }), 
+                delay, TimeUnit.MILLISECONDS);
+        });
+        return cf;
+    }
+
     protected DeleteObjectsAccumulator newDeleteObjectsAccumulator() {
         return new DeleteObjectsAccumulator(this::doDeleteObjects);
     }

s3stream/src/main/java/com/automq/stream/s3/operator/AwsObjectStorage.java

@@ -57,6 +57,7 @@
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
 import software.amazon.awssdk.core.exception.ApiCallAttemptTimeoutException;
 import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.core.retry.RetryMode;
 import software.amazon.awssdk.http.HttpStatusCode;
 import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
 import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;
@@ -309,21 +310,23 @@ public CompletableFuture<Void> doDeleteObjects(List<String> objectKeys) {
             .delete(Delete.builder().objects(toDeleteKeys).build())
             .build();
 
-        CompletableFuture<Void> cf = new CompletableFuture<>();
-        this.writeS3Client.deleteObjects(request)
-            .thenAccept(resp -> {
-                try {
-                    checkDeleteObjectsResponse(resp);
-                    cf.complete(null);
-                } catch (Throwable ex) {
+        return retryOnThrottle(() -> {
+            CompletableFuture<Void> cf = new CompletableFuture<>();
+            writeS3Client.deleteObjects(request)
+                .thenAccept(resp -> {
+                    try {
+                        checkDeleteObjectsResponse(resp);
+                        cf.complete(null);
+                    } catch (Throwable ex) {
+                        cf.completeExceptionally(ex);
+                    }
+                })
+                .exceptionally(ex -> {
                     cf.completeExceptionally(ex);
-                }
-            })
-            .exceptionally(ex -> {
-                cf.completeExceptionally(ex);
-                return null;
-            });
-        return cf;
+                    return null;
+                });
+            return cf;
+        }, S3Operation.DELETE_OBJECTS, 0);
     }
 
     @Override
@@ -438,6 +441,7 @@ protected ClientOverrideConfiguration clientOverrideConfiguration(long apiCallTi
         return ClientOverrideConfiguration.builder()
             .apiCallTimeout(Duration.ofMillis(apiCallTimeoutMs))
             .apiCallAttemptTimeout(Duration.ofMillis(apiCallAttemptTimeoutMs))
+            .retryStrategy(RetryMode.STANDARD)
             .build();
     }