feat: support 'prefetch' in async pagination

Author: bevzzz

src/it/java/io/weaviate/integration/PaginationITest.java

@@ -143,7 +143,7 @@ public void testAsyncPaginator() throws IOException, InterruptedException, Execu
 
       // Act
       var objectCount = new AtomicInteger();
-      var countAll = things.paginate(p -> p.pageSize(5))
+      var countAll = things.paginate(p -> p.pageSize(5).prefetch(true))
           .forEach(__ -> objectCount.getAndIncrement());
 
       // Assert

src/main/java/io/weaviate/client6/v1/api/collections/pagination/AsyncPaginator.java

@@ -36,12 +36,12 @@ public AsyncPaginator(Builder<PropertiesT> builder) {
           return this.query.fetchObjects(fn).thenApply(QueryResponse::objects);
         });
 
-    this.resultSet = CompletableFuture.completedFuture(rs);
+    this.resultSet = builder.prefetch ? rs.fetchNextPage() : CompletableFuture.completedFuture(rs);
   }
 
   public CompletableFuture<Void> forEach(Consumer<WeaviateObject<PropertiesT, Object, QueryMetadata>> action) {
     return resultSet
-        .thenCompose(AsyncResultSet::fetchNextPage)
+        .thenCompose(rs -> rs.isEmpty() ? rs.fetchNextPage() : CompletableFuture.completedFuture(rs))
         .thenCompose(processPageAndAdvance(action));
   }