fix(s3stream): add pending requests await timeout for S3Stream#close (#2750)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

s3stream/src/main/java/com/automq/stream/s3/S3Stream.java

@@ -393,10 +393,11 @@ public CompletableFuture<Void> close(boolean force) {
 
             // await all pending append/fetch/trim request
             List<CompletableFuture<?>> pendingRequests = new ArrayList<>(pendingAppends);
+            // add timeout to prevent the fetch(catch-up read) network throttle to block Stream#close.
             if (GlobalSwitch.STRICT) {
-                pendingRequests.addAll(pendingFetches);
+                pendingRequests.addAll(FutureUtil.timeoutAndSilence(pendingFetches.stream(), 10, TimeUnit.SECONDS));
             }
-            pendingRequests.add(lastPendingTrim);
+            pendingRequests.add(FutureUtil.timeoutAndSilence(lastPendingTrim, 10, TimeUnit.SECONDS));
             if (force) {
                 pendingRequests.forEach(cf -> cf.completeExceptionally(new StreamClientException(ErrorCode.UNEXPECTED, "FORCE_CLOSE")));
             }

s3stream/src/main/java/com/automq/stream/utils/FutureUtil.java

@@ -23,6 +23,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Iterator;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
@@ -31,6 +32,8 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.Timeout;
@@ -165,4 +168,11 @@ public static <T> CompletableFuture<T> timeoutWithNewReturn(CompletableFuture<T>
         return newCf;
     }
 
+    public static List<CompletableFuture<?>> timeoutAndSilence(Stream<CompletableFuture<?>> stream, long timeout, TimeUnit timeUnit) {
+        return stream.map(l -> timeoutAndSilence(l, timeout, timeUnit)).collect(Collectors.toList());
+    }
+
+    public static <T> CompletableFuture<T> timeoutAndSilence(CompletableFuture<T> cf, long timeout, TimeUnit timeUnit) {
+        return cf.orTimeout(timeout, timeUnit).exceptionally(ex -> null);
+    }
 }