feat: add additional information for exceptions during pagination

Author: bevzzz

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

@@ -6,6 +6,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -18,6 +19,7 @@
 import io.weaviate.client6.v1.api.collections.Property;
 import io.weaviate.client6.v1.api.collections.WeaviateMetadata;
 import io.weaviate.client6.v1.api.collections.WeaviateObject;
+import io.weaviate.client6.v1.api.collections.pagination.WeaviatePaginationException;
 import io.weaviate.containers.Container;
 
 public class PaginationITest extends ConcurrentTest {
@@ -89,7 +91,7 @@ public void testResumePagination() throws IOException {
         .reduce((prev, next) -> next).get();
 
     // Act
-    var remaining = things.paginate(p -> p.resumeFrom(lastId)).stream().count();
+    var remaining = things.paginate(p -> p.fromCursor(lastId)).stream().count();
 
     // Assert
     Assertions.assertThat(remaining).isEqualTo(5);
@@ -157,4 +159,34 @@ public void testAsyncPaginator() throws IOException, InterruptedException, Execu
           .isEqualTo(count);
     }
   }
+
+  @Test(expected = WeaviatePaginationException.class)
+  public void testFailedPagination() throws IOException {
+    var things = client.collections.use("Unknown");
+    things.paginate().forEach(System.out::println);
+  }
+
+  @Test(expected = WeaviatePaginationException.class)
+  public void testFailedAsyncPagination_forEach() throws Throwable {
+    try (final var async = client.async()) {
+      var things = async.collections.use("Unknown");
+      try {
+        things.paginate().forEach(__ -> System.out.println("called once")).join();
+      } catch (CompletionException e) {
+        throw e.getCause(); // CompletableFuture exceptions are always wrapped
+      }
+    }
+  }
+
+  @Test(expected = WeaviatePaginationException.class)
+  public void testFailedAsyncPagination_forPage() throws Throwable {
+    try (final var async = client.async()) {
+      var things = async.collections.use("Unknown");
+      try {
+        things.paginate().forPage(__ -> System.out.println("called once")).join();
+      } catch (CompletionException e) {
+        throw e.getCause(); // CompletableFuture exceptions are always wrapped
+      }
+    }
+  }
 }

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

@@ -6,6 +6,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
 
 import org.assertj.core.api.Assertions;
@@ -402,15 +403,15 @@ public void testBadRequest_async() throws Throwable {
           collection -> collection
               .properties(Property.text("name"))
               .vectors(Vectorizers.text2vecContextionary()))
-          .get();
+          .join();
 
       var things = async.collections.use(nsThings);
-      var balloon = things.data.insert(Map.of("name", "balloon")).get();
+      var balloon = things.data.insert(Map.of("name", "balloon")).join();
 
       try {
-        things.query.nearObject(balloon.uuid(), q -> q.limit(-1)).get();
-      } catch (ExecutionException e) {
-        throw e.getCause();
+        things.query.nearObject(balloon.uuid(), q -> q.limit(-1)).join();
+      } catch (CompletionException e) {
+        throw e.getCause(); // CompletableFuture exceptions are always wrapped
       }
     }
   }

src/main/java/io/weaviate/client6/v1/api/WeaviateApiException.java

@@ -9,8 +9,8 @@ public class WeaviateApiException extends RuntimeException {
   private final String errorMessage;
   private final Source source;
   private final String endpoint;
-  private final Integer statusCode;
-  private final String grpcStatus;
+  private final Integer httpStatusCode;
+  private final io.grpc.Status.Code grpcStatusCode;
 
   private enum Source {
     HTTP, GRPC;
@@ -22,32 +22,48 @@ public static WeaviateApiException http(String method, String endpoint, int stat
 
   public static WeaviateApiException gRPC(io.grpc.StatusRuntimeException ex) {
     var status = ex.getStatus();
-    return new WeaviateApiException(status.getCode().toString(), status.getDescription());
+    return new WeaviateApiException(status.getCode(), status.getDescription());
   }
 
-  private WeaviateApiException(String status, String errorMessage) {
-    super("%s: %s".formatted(status, errorMessage));
+  private WeaviateApiException(io.grpc.Status.Code code, String errorMessage) {
+    super("%s: %s".formatted(code, errorMessage));
     this.source = Source.GRPC;
     this.errorMessage = errorMessage;
-    this.grpcStatus = status;
+    this.grpcStatusCode = code;
     this.endpoint = null;
-    this.statusCode = null;
+    this.httpStatusCode = null;
   }
 
   private WeaviateApiException(String method, String endpoint, int statusCode, String errorMessage) {
     super("HTTP %d: %s %s: %s".formatted(statusCode, method, endpoint, errorMessage));
     this.source = Source.HTTP;
     this.errorMessage = errorMessage;
     this.endpoint = endpoint;
-    this.statusCode = statusCode;
-    this.grpcStatus = null;
+    this.httpStatusCode = statusCode;
+    this.grpcStatusCode = null;
+  }
+
+  public boolean isGPRC() {
+    return source == Source.GRPC;
+  }
+
+  public String grpcStatusCode() {
+    return grpcStatusCode.toString();
+  }
+
+  public boolean isHTTP() {
+    return source == Source.HTTP;
   }
 
   public String endpoint() {
     return endpoint;
   }
 
-  public Integer statusCode() {
-    return statusCode;
+  public Integer httpStatusCode() {
+    return httpStatusCode;
+  }
+
+  public String getError() {
+    return errorMessage;
   }
 }

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

@@ -40,6 +40,15 @@ public boolean isEmpty() {
     return this.currentPage.isEmpty();
   }
 
+  /**
+   * Fetch an {@link AsyncPage} containing the next {@code pageSize} results
+   * and advance the cursor.
+   *
+   * <p>
+   * The returned stage may complete exceptionally in case the underlying
+   * query fails. Callers are advised to use exception-aware
+   * {@link CompletableFuture#handle} to process page results.
+   */
   public CompletableFuture<AsyncPage<PropertiesT>> fetchNextPage() {
     return fetch.apply(cursor, pageSize)
         .thenApply(nextPage -> {

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

@@ -31,9 +31,16 @@ public AsyncPaginator(Builder<PropertiesT> builder) {
     var rs = new AsyncPage<PropertiesT>(
         cursor,
         pageSize,
-        (after, limit) -> {
-          var fn = ObjectBuilder.partial(queryOptions, q -> q.after(after).limit(limit));
-          return this.query.fetchObjects(fn).thenApply(QueryResponse::objects);
+        (cursor, pageSize) -> {
+          var fn = ObjectBuilder.partial(queryOptions, q -> q.after(cursor).limit(pageSize));
+          return this.query.fetchObjects(fn)
+              .handle((response, ex) -> {
+                if (ex != null) {
+                  throw WeaviatePaginationException.after(cursor, pageSize, ex);
+                }
+                return response;
+              })
+              .thenApply(QueryResponse::objects);
         });
 
     this.resultSet = builder.prefetch ? rs.fetchNextPage() : CompletableFuture.completedFuture(rs);
@@ -51,17 +58,17 @@ public CompletableFuture<Void> forPage(Consumer<List<WeaviateObject<PropertiesT,
         .thenCompose(processPageAndAdvance(action));
   }
 
-  public Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processEachAndAdvance(
+  private static <PropertiesT> Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processEachAndAdvance(
       Consumer<WeaviateObject<PropertiesT, Object, QueryMetadata>> action) {
     return processAndAdvanceFunc(rs -> rs.forEach(action));
   }
 
-  public Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processPageAndAdvance(
+  private static <PropertiesT> Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processPageAndAdvance(
       Consumer<List<WeaviateObject<PropertiesT, Object, QueryMetadata>>> action) {
     return processAndAdvanceFunc(rs -> action.accept(rs.items()));
   }
 
-  public Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processAndAdvanceFunc(
+  private static <PropertiesT> Function<AsyncPage<PropertiesT>, CompletableFuture<Void>> processAndAdvanceFunc(
       Consumer<AsyncPage<PropertiesT>> action) {
     return rs -> {
       // Empty result set means there were no more objects to fetch.
@@ -105,11 +112,17 @@ public Builder<PropertiesT> pageSize(int pageSize) {
       return this;
     }
 
-    public Builder<PropertiesT> resumeFrom(String uuid) {
+    /** Set a cursor (object UUID) to start pagination from. */
+    public Builder<PropertiesT> fromCursor(String uuid) {
       this.cursor = uuid;
       return this;
     }
 
+    /**
+     * When prefetch is enabled, the first page is retrieved before any of the
+     * terminating methods ({@link AsyncPaginator#forEach},
+     * {@link AsyncPaginator#forPage}) are called on the paginator.
+     */
     public Builder<PropertiesT> prefetch(boolean enable) {
       this.prefetch = enable;
       return this;

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

@@ -37,7 +37,11 @@ public Spliterator<WeaviateObject<PropertiesT, Object, QueryMetadata>> spliterat
     return new CursorSpliterator<PropertiesT>(cursor, pageSize,
         (after, limit) -> {
           var fn = ObjectBuilder.partial(queryOptions, q -> q.after(after).limit(limit));
-          return query.fetchObjects(fn).objects();
+          try {
+            return query.fetchObjects(fn).objects();
+          } catch (Exception e) {
+            throw WeaviatePaginationException.after(cursor, pageSize, e);
+          }
         });
   }
 
@@ -75,7 +79,8 @@ public Builder<T> pageSize(int pageSize) {
       return this;
     }
 
-    public Builder<T> resumeFrom(String uuid) {
+    /** Set a cursor (object UUID) to start pagination from. */
+    public Builder<T> fromCursor(String uuid) {
       this.cursor = uuid;
       return this;
     }