fix: support list more than 1000 objects by prefix (#2660) (#2666)

This commit fixes an issue where the doList method in AwsObjectStorage.java
did not handle paginated results from the S3 listObjectsV2 API. The
method now recursively fetches all pages of objects, ensuring that it can
retrieve more than the default 1000 objects.

Author: Gezi-lzq

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

@@ -362,12 +362,33 @@ void doClose() {
 
     @Override
     CompletableFuture<List<ObjectInfo>> doList(String prefix) {
-        return readS3Client.listObjectsV2(builder -> builder.bucket(bucket).prefix(prefix))
-            .thenApply(resp ->
-                resp.contents()
-                    .stream()
-                    .map(object -> new ObjectInfo(bucketURI.bucketId(), object.key(), object.lastModified().toEpochMilli(), object.size()))
-                    .collect(Collectors.toList()));
+        CompletableFuture<List<ObjectInfo>> resultFuture = new CompletableFuture<>();
+        List<ObjectInfo> allObjects = new ArrayList<>();
+        listNextBatch(prefix, null, allObjects, resultFuture);
+        return resultFuture;
+    }
+
+    private void listNextBatch(String prefix, String continuationToken, List<ObjectInfo> allObjects,
+        CompletableFuture<List<ObjectInfo>> resultFuture) {
+        readS3Client.listObjectsV2(builder -> {
+            builder.bucket(bucket).prefix(prefix);
+            if (continuationToken != null) {
+                builder.continuationToken(continuationToken);
+            }
+        }).thenAccept(resp -> {
+            resp.contents()
+                .stream()
+                .map(object -> new ObjectInfo(bucketURI.bucketId(), object.key(), object.lastModified().toEpochMilli(), object.size()))
+                .forEach(allObjects::add);
+            if (resp.isTruncated()) {
+                listNextBatch(prefix, resp.nextContinuationToken(), allObjects, resultFuture);
+            } else {
+                resultFuture.complete(allObjects);
+            }
+        }).exceptionally(ex -> {
+            resultFuture.completeExceptionally(ex);
+            return null;
+        });
     }
 
     @Override