From f4f79a3d92ea345210b216d9883360499b1fe4fa Mon Sep 17 00:00:00 2001 From: SoulPancake Date: Sat, 29 Aug 2026 01:01:10 +0530 Subject: [PATCH 1/3] fix: remove per-call thread pools in batchCheck and clientBatchCheck Replace the per-call ScheduledThreadPoolExecutor and blocking CountDownLatch with CompletableFuture lane composition. maxParallelRequests now actually bounds in-flight requests (it previously only sized a pool of threads that fired all requests at once), callers are no longer blocked until completion, non-positive maxParallelRequests is rejected, and response-processing errors fail the returned future instead of being swallowed while remaining batches are still attempted. --- .../openfga/sdk/api/client/OpenFgaClient.java | 125 ++++--- .../sdk/api/client/OpenFgaClientTest.java | 305 ++++++++++++++++-- 2 files changed, 351 insertions(+), 79 deletions(-) diff --git a/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java b/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java index daf96f11..33f71736 100644 --- a/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java +++ b/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java @@ -19,6 +19,7 @@ import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; +import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.Stream; @@ -734,6 +735,35 @@ private Stream> chunksOf(int chunkSize, List list) { return chunks.build(); } + /** + * Runs one asynchronous task per item with at most {@code maxParallelism} tasks in flight at + * a time, without creating any threads or executors. Items are distributed round-robin across + * lanes; each lane executes its items sequentially through future composition while all lanes + * proceed concurrently. Tasks are expected to capture their own errors and complete normally; + * a task that completes exceptionally fails the returned future and skips the remaining items + * in its lane. + * + * @return a future that completes when every task has completed + * @throws IllegalArgumentException when {@code maxParallelism} is not positive + */ + private static CompletableFuture executeInLanes( + List items, Function> task, int maxParallelism) { + if (maxParallelism <= 0) { + throw new IllegalArgumentException("maxParallelism must be a positive integer"); + } + int lanes = Math.min(maxParallelism, items.size()); + CompletableFuture[] laneFutures = new CompletableFuture[lanes]; + for (int lane = 0; lane < lanes; lane++) { + CompletableFuture chain = CompletableFuture.completedFuture(null); + for (int i = lane; i < items.size(); i += lanes) { + var item = items.get(i); + chain = chain.thenCompose(unused -> task.apply(item)); + } + laneFutures[lane] = chain; + } + return CompletableFuture.allOf(laneFutures); + } + /** * WriteTuples - Utility method to write tuples, wraps Write * @@ -901,28 +931,20 @@ public CompletableFuture> clientBatchCheck( int maxParallelRequests = options.getMaxParallelRequests() != null ? options.getMaxParallelRequests() : FgaConstants.CLIENT_MAX_METHOD_PARALLEL_REQUESTS; - var executor = Executors.newScheduledThreadPool(maxParallelRequests); - var latch = new CountDownLatch(requests.size()); - + if (maxParallelRequests <= 0) { + throw new FgaInvalidParameterException("maxParallelRequests", "ClientBatchCheck"); + } var responses = new ConcurrentLinkedQueue(); final var clientCheckOptions = options.asClientCheckOptions(); - Consumer singleClientCheckRequest = + Function> singleClientCheckRequest = request -> call(() -> this.check(request, clientCheckOptions)) - .handleAsync(ClientBatchCheckClientResponse.asyncHandler(request)) - .thenAccept(responses::add) - .thenRun(latch::countDown); + .handle(ClientBatchCheckClientResponse.asyncHandler(request)) + .thenAccept(responses::add); - try { - requests.forEach(request -> executor.execute(() -> singleClientCheckRequest.accept(request))); - latch.await(); - return CompletableFuture.completedFuture(new ArrayList<>(responses)); - } catch (Exception e) { - return CompletableFuture.failedFuture(e); - } finally { - executor.shutdown(); - } + return executeInLanes(requests, singleClientCheckRequest, maxParallelRequests) + .thenApply(unused -> new ArrayList<>(responses)); } /** @@ -1002,15 +1024,15 @@ public CompletableFuture batchCheck( int maxParallelRequests = options.getMaxParallelRequests() != null ? options.getMaxParallelRequests() : FgaConstants.CLIENT_MAX_METHOD_PARALLEL_REQUESTS; - var executor = Executors.newScheduledThreadPool(maxParallelRequests); - var latch = new CountDownLatch(batchedChecks.size()); - + if (maxParallelRequests <= 0) { + throw new FgaInvalidParameterException("maxParallelRequests", "BatchCheck"); + } var responses = new ConcurrentLinkedQueue(); var failure = new AtomicReference(); var override = new ConfigurationOverride().addHeaders(options); - Consumer> singleBatchCheckRequest = request -> call(() -> { + Function, CompletableFuture> singleBatchCheckRequest = request -> call(() -> { BatchCheckRequest body = new BatchCheckRequest().checks(request); if (options.getConsistency() != null) { body.consistency(options.getConsistency()); @@ -1027,42 +1049,39 @@ public CompletableFuture batchCheck( return api.batchCheck(configuration.getStoreId(), body, override); }) - .whenComplete((batchCheckResponseApiResponse, throwable) -> { - try { - if (throwable != null) { - failure.compareAndSet(null, throwable); - return; - } - - Map response = - batchCheckResponseApiResponse.getData().getResult(); - - List batchResults = new ArrayList<>(); - response.forEach((key, result) -> { - boolean allowed = Boolean.TRUE.equals(result.getAllowed()); - ClientBatchCheckItem checkItem = correlationIdToCheck.get(key); - var singleResponse = - new ClientBatchCheckSingleResponse(allowed, checkItem, key, result.getError()); - batchResults.add(singleResponse); - }); - responses.addAll(batchResults); - } finally { - latch.countDown(); + .handle((batchCheckResponseApiResponse, throwable) -> { + if (throwable != null) { + failure.compareAndSet(null, throwable); + return null; } + + Map response = + batchCheckResponseApiResponse.getData().getResult(); + + List batchResults = new ArrayList<>(); + response.forEach((key, result) -> { + boolean allowed = Boolean.TRUE.equals(result.getAllowed()); + ClientBatchCheckItem checkItem = correlationIdToCheck.get(key); + var singleResponse = + new ClientBatchCheckSingleResponse(allowed, checkItem, key, result.getError()); + batchResults.add(singleResponse); + }); + responses.addAll(batchResults); + return null; + }) + .exceptionally(processingFailure -> { + failure.compareAndSet(null, processingFailure); + return null; }); - try { - batchedChecks.forEach(batch -> executor.execute(() -> singleBatchCheckRequest.accept(batch))); - latch.await(); - if (failure.get() != null) { - return CompletableFuture.failedFuture(failure.get()); - } - return CompletableFuture.completedFuture(new ClientBatchCheckResponse(new ArrayList<>(responses))); - } catch (Exception e) { - return CompletableFuture.failedFuture(e); - } finally { - executor.shutdown(); - } + return executeInLanes(batchedChecks, singleBatchCheckRequest, maxParallelRequests) + .thenCompose(unused -> { + Throwable batchFailure = failure.get(); + if (batchFailure != null) { + return CompletableFuture.failedFuture(batchFailure); + } + return CompletableFuture.completedFuture(new ClientBatchCheckResponse(new ArrayList<>(responses))); + }); } /** diff --git a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java index bad5b5fd..b1185d1c 100644 --- a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java +++ b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java @@ -18,21 +18,29 @@ import dev.openfga.sdk.api.model.*; import dev.openfga.sdk.constants.FgaConstants; import dev.openfga.sdk.errors.*; +import java.net.URI; import java.net.http.HttpClient; +import java.net.http.HttpHeaders; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; import java.time.Duration; import java.time.OffsetDateTime; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.IntStream; import java.util.stream.Stream; +import javax.net.ssl.SSLSession; import org.hamcrest.BaseMatcher; import org.hamcrest.Description; import org.hamcrest.Matcher; @@ -1964,37 +1972,282 @@ public void clientBatchCheck() throws Exception { } @Test - public void shouldShutdownExecutorAfterBatchCheck() throws Exception { + public void batchCheckMethodsShouldNotCreateThreadPools() throws Exception { // Given - ScheduledExecutorService mockExecutor = mock(ScheduledExecutorService.class); - - try (MockedStatic mockedExecutors = mockStatic(Executors.class)) { - mockedExecutors - .when(() -> Executors.newScheduledThreadPool(anyInt())) - .thenReturn(mockExecutor); - - // mockExecutor needs to handle tasks submitted to it so latch can count down - doAnswer(invocation -> { - Runnable task = invocation.getArgument(0); - task.run(); - return null; - }) - .when(mockExecutor) - .execute(any(Runnable.class)); + String checkUrl = String.format("%s/stores/%s/check", FgaConstants.TEST_API_URL, DEFAULT_STORE_ID); + String batchCheckUrl = String.format("%s/stores/%s/batch-check", FgaConstants.TEST_API_URL, DEFAULT_STORE_ID); + mockHttpClient.onPost(checkUrl).doReturn(200, "{\"allowed\":true}"); + mockHttpClient + .onPost(batchCheckUrl) + .doReturn(200, "{\"result\": {\"cor-1\": {\"allowed\": true, \"error\": null}}}"); - ClientCheckRequest request = new ClientCheckRequest() - ._object(DEFAULT_OBJECT) - .relation(DEFAULT_RELATION) - .user(DEFAULT_USER); - ClientBatchCheckClientOptions options = new ClientBatchCheckClientOptions() - .authorizationModelId(DEFAULT_AUTH_MODEL_ID) - .consistency(ConsistencyPreference.MINIMIZE_LATENCY); + ClientCheckRequest checkRequest = new ClientCheckRequest() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER); + ClientBatchCheckItem batchItem = new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-1"); + try (MockedStatic mockedExecutors = mockStatic(Executors.class, CALLS_REAL_METHODS)) { // When - fga.clientBatchCheck(List.of(request), options).get(); + fga.clientBatchCheck(List.of(checkRequest)).get(); + fga.batchCheck(new ClientBatchCheckRequest().checks(List.of(batchItem))) + .get(); // Then - verify(mockExecutor).shutdown(); + mockedExecutors.verify(() -> Executors.newScheduledThreadPool(anyInt()), never()); + } + } + + @Test + public void clientBatchCheckHonorsMaxParallelRequests() throws Exception { + // Given + var pending = new CopyOnWriteArrayList>>(); + var fga = clientBackedByPendingResponses(pending); + List requests = IntStream.range(0, 5) + .mapToObj(ignored -> new ClientCheckRequest() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER)) + .collect(Collectors.toList()); + var options = new ClientBatchCheckClientOptions().maxParallelRequests(2); + + // When + var future = assertTimeoutPreemptively(Duration.ofSeconds(5), () -> fga.clientBatchCheck(requests, options)); + + // Then + assertFalse(future.isDone()); + Thread.sleep(200); + assertEquals(2, pending.size()); + + // completing one response frees its lane to start the next queued check + pending.get(0).complete(fakeResponse("{\"allowed\":true}")); + awaitSize(pending, 3); + assertFalse(future.isDone()); + + completeAllUntilDone(future, pending, "{\"allowed\":true}"); + assertEquals(5, future.get(5, TimeUnit.SECONDS).size()); + assertEquals(5, pending.size()); + } + + @Test + public void batchCheckHonorsMaxParallelRequests() throws Exception { + // Given: 120 checks form three sub-batches with the default maxBatchSize of 50 + var pending = new CopyOnWriteArrayList>>(); + var fga = clientBackedByPendingResponses(pending); + List checks = IntStream.range(0, 120) + .mapToObj(i -> new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-" + i)) + .collect(Collectors.toList()); + var options = new ClientBatchCheckOptions().maxParallelRequests(1); + + // When + var future = assertTimeoutPreemptively( + Duration.ofSeconds(5), () -> fga.batchCheck(new ClientBatchCheckRequest().checks(checks), options)); + + // Then: sub-batches are sent strictly one at a time + assertFalse(future.isDone()); + Thread.sleep(200); + assertEquals(1, pending.size()); + + pending.get(0).complete(fakeResponse("{\"result\": {}}")); + awaitSize(pending, 2); + + pending.get(1).complete(fakeResponse("{\"result\": {}}")); + awaitSize(pending, 3); + assertFalse(future.isDone()); + + pending.get(2).complete(fakeResponse("{\"result\": {}}")); + future.get(5, TimeUnit.SECONDS); + assertEquals(3, pending.size()); + } + + @Test + public void batchCheckMethodsRejectNonPositiveMaxParallelRequests() { + // Given + ClientCheckRequest checkRequest = new ClientCheckRequest() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER); + ClientBatchCheckItem batchItem = new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-1"); + + for (int invalid : new int[] {0, -1}) { + var clientOptions = new ClientBatchCheckClientOptions().maxParallelRequests(invalid); + assertThrows( + FgaInvalidParameterException.class, + () -> fga.clientBatchCheck(List.of(checkRequest), clientOptions)); + + var serverOptions = new ClientBatchCheckOptions().maxParallelRequests(invalid); + assertThrows( + FgaInvalidParameterException.class, + () -> fga.batchCheck(new ClientBatchCheckRequest().checks(List.of(batchItem)), serverOptions)); + } + } + + @Test + public void batchCheckFailsWhenResponseProcessingFails() throws Exception { + // Given: a 200 response whose body is missing the result payload + String batchCheckUrl = String.format("%s/stores/%s/batch-check", FgaConstants.TEST_API_URL, DEFAULT_STORE_ID); + mockHttpClient.onPost(batchCheckUrl).doReturn(200, "{\"result\": null}"); + + ClientBatchCheckItem batchItem = new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-1"); + + // When + var future = fga.batchCheck(new ClientBatchCheckRequest().checks(List.of(batchItem))); + + // Then: the processing error fails the returned future instead of being dropped + var exception = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertNotNull(exception.getCause()); + } + + @Test + public void batchCheckAttemptsRemainingBatchesWhenResponseProcessingFails() throws Exception { + // Given: 60 checks form two sub-batches with the default maxBatchSize of 50 + var pending = new CopyOnWriteArrayList>>(); + var fga = clientBackedByPendingResponses(pending); + List checks = IntStream.range(0, 60) + .mapToObj(i -> new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-" + i)) + .collect(Collectors.toList()); + var options = new ClientBatchCheckOptions().maxParallelRequests(1); + + // When + var future = assertTimeoutPreemptively( + Duration.ofSeconds(5), () -> fga.batchCheck(new ClientBatchCheckRequest().checks(checks), options)); + + assertFalse(future.isDone()); + Thread.sleep(200); + assertEquals(1, pending.size()); + + // the first sub-batch returns a malformed 200 (null result) that fails during processing + pending.get(0).complete(fakeResponse("{\"result\": null}")); + + // the lane must still send the remaining sub-batch instead of aborting early + awaitSize(pending, 2); + pending.get(1).complete(fakeResponse("{\"result\": {}}")); + + // Then + var exception = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertNotNull(exception.getCause()); + } + + @Test + public void clientBatchCheckWithEmptyInputCompletesWithoutDispatch() throws Exception { + // Given + var pending = new CopyOnWriteArrayList>>(); + var fga = clientBackedByPendingResponses(pending); + + // When + var results = fga.clientBatchCheck(List.of()).get(5, TimeUnit.SECONDS); + + // Then + assertTrue(results.isEmpty()); + assertEquals(0, pending.size()); + } + + private OpenFgaClient clientBackedByPendingResponses(List>> pending) + throws Exception { + HttpClient pendingClient = mock(HttpClient.class); + doAnswer(invocation -> { + var responseFuture = new CompletableFuture>(); + pending.add(responseFuture); + return responseFuture; + }) + .when(pendingClient) + .sendAsync(any(), any()); + + var builder = mock(HttpClient.Builder.class); + when(builder.executor(any())).thenReturn(builder); + when(builder.connectTimeout(any())).thenReturn(builder); + when(builder.build()).thenReturn(pendingClient); + return new OpenFgaClient(clientConfiguration, new ApiClient(builder, new ObjectMapper())); + } + + private static void awaitSize(List list, int expected) throws InterruptedException { + long deadline = System.currentTimeMillis() + 5000; + while (list.size() < expected && System.currentTimeMillis() < deadline) { + Thread.sleep(10); + } + assertEquals(expected, list.size()); + } + + private static void completeAllUntilDone( + CompletableFuture future, List>> pending, String body) + throws InterruptedException { + long deadline = System.currentTimeMillis() + 5000; + while (!future.isDone() && System.currentTimeMillis() < deadline) { + pending.forEach(responseFuture -> responseFuture.complete(fakeResponse(body))); + Thread.sleep(10); + } + assertTrue(future.isDone()); + } + + private static HttpResponse fakeResponse(String body) { + return new FakeHttpResponse(body); + } + + private static final class FakeHttpResponse implements HttpResponse { + private final String body; + + private FakeHttpResponse(String body) { + this.body = body; + } + + @Override + public int statusCode() { + return 200; + } + + @Override + public HttpRequest request() { + return null; + } + + @Override + public Optional> previousResponse() { + return Optional.empty(); + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of(Map.of(), (name, value) -> true); + } + + @Override + public String body() { + return body; + } + + @Override + public Optional sslSession() { + return Optional.empty(); + } + + @Override + public URI uri() { + return URI.create(FgaConstants.TEST_API_URL); + } + + @Override + public HttpClient.Version version() { + return HttpClient.Version.HTTP_1_1; } } From 400be88f5d03c0453b4680d21b07c92bc9852216 Mon Sep 17 00:00:00 2001 From: SoulPancake Date: Sat, 29 Aug 2026 12:19:17 +0530 Subject: [PATCH 2/3] fix: reject non-positive maxBatchSize in batchCheck A zero maxBatchSize made the sub-batch math throw ArithmeticException and a negative one silently dropped every check, returning an empty success. Validate it like maxParallelRequests. --- .../openfga/sdk/api/client/OpenFgaClient.java | 3 +++ .../sdk/api/client/OpenFgaClientTest.java | 17 +++++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java b/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java index 33f71736..28e74de6 100644 --- a/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java +++ b/src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java @@ -1016,6 +1016,9 @@ public CompletableFuture batchCheck( int maxBatchSize = options.getMaxBatchSize() != null ? options.getMaxBatchSize() : FgaConstants.CLIENT_MAX_BATCH_SIZE; + if (maxBatchSize <= 0) { + throw new FgaInvalidParameterException("maxBatchSize", "BatchCheck"); + } List> batchedChecks = IntStream.range( 0, (collect.size() + maxBatchSize - 1) / maxBatchSize) .mapToObj(i -> collect.subList(i * maxBatchSize, Math.min((i + 1) * maxBatchSize, collect.size()))) diff --git a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java index b1185d1c..f8a0c432 100644 --- a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java +++ b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java @@ -2094,6 +2094,23 @@ public void batchCheckMethodsRejectNonPositiveMaxParallelRequests() { } } + @Test + public void batchCheckRejectsNonPositiveMaxBatchSize() { + // Given + ClientBatchCheckItem batchItem = new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-1"); + + for (int invalid : new int[] {0, -1}) { + var options = new ClientBatchCheckOptions().maxBatchSize(invalid); + assertThrows( + FgaInvalidParameterException.class, + () -> fga.batchCheck(new ClientBatchCheckRequest().checks(List.of(batchItem)), options)); + } + } + @Test public void batchCheckFailsWhenResponseProcessingFails() throws Exception { // Given: a 200 response whose body is missing the result payload From f4b7f44fed28a7c747dae3d4b478ecb3322b70f2 Mon Sep 17 00:00:00 2001 From: SoulPancake Date: Mon, 31 Aug 2026 18:58:57 +0530 Subject: [PATCH 3/3] test: cover sibling-lane isolation when a batchCheck sub-batch fails processing With maxParallelRequests(2) and four sub-batches distributed round-robin across two lanes, a processing failure in one lane's sub-batch must not stop the sibling lane: both lanes keep dispatching their remaining sub-batches, all four are attempted, and the processing error still fails the returned future. The existing coverage only proved same-lane continuation with maxParallelRequests(1). --- .../sdk/api/client/OpenFgaClientTest.java | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java index f8a0c432..a5eefe4f 100644 --- a/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java +++ b/src/test/java/dev/openfga/sdk/api/client/OpenFgaClientTest.java @@ -2165,6 +2165,46 @@ public void batchCheckAttemptsRemainingBatchesWhenResponseProcessingFails() thro assertNotNull(exception.getCause()); } + @Test + public void batchCheckContinuesSiblingLaneWhenResponseProcessingFails() throws Exception { + // Given: 200 checks form four sub-batches with the default maxBatchSize of 50, distributed + // round-robin across two lanes (lane 0: sub-batches 0 and 2, lane 1: sub-batches 1 and 3) + var pending = new CopyOnWriteArrayList>>(); + var fga = clientBackedByPendingResponses(pending); + List checks = IntStream.range(0, 200) + .mapToObj(i -> new ClientBatchCheckItem() + ._object(DEFAULT_OBJECT) + .relation(DEFAULT_RELATION) + .user(DEFAULT_USER) + .correlationId("cor-" + i)) + .collect(Collectors.toList()); + var options = new ClientBatchCheckOptions().maxParallelRequests(2); + + // When: both lanes dispatch their first sub-batch concurrently + var future = assertTimeoutPreemptively( + Duration.ofSeconds(5), () -> fga.batchCheck(new ClientBatchCheckRequest().checks(checks), options)); + awaitSize(pending, 2); + + // lane 0's sub-batch returns a malformed 200 (null result) that fails during processing + pending.get(0).complete(fakeResponse("{\"result\": null}")); + + // the failed lane still dispatches its remaining sub-batch + awaitSize(pending, 3); + + // and the healthy sibling lane is unaffected: completing its first sub-batch dispatches its next one + pending.get(1).complete(fakeResponse("{\"result\": {}}")); + awaitSize(pending, 4); + assertFalse(future.isDone()); + + pending.get(2).complete(fakeResponse("{\"result\": {}}")); + pending.get(3).complete(fakeResponse("{\"result\": {}}")); + + // Then: every sub-batch was attempted and the processing error still fails the returned future + var exception = assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS)); + assertNotNull(exception.getCause()); + assertEquals(4, pending.size()); + } + @Test public void clientBatchCheckWithEmptyInputCompletesWithoutDispatch() throws Exception { // Given