Skip to content

fix: remove per-call thread pools in batchCheck and clientBatchCheck #379

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SoulPancake wants to merge 4 commits into
base: main
Choose a base branch
from
+411 −79
Open

fix: remove per-call thread pools in batchCheck and clientBatchCheck #379

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f4f79a3
fix: remove per-call thread pools in batchCheck and clientBatchCheck
SoulPancake Aug 28, 2026
400be88
fix: reject non-positive maxBatchSize in batchCheck
SoulPancake Aug 29, 2026
f4b7f44
test: cover sibling-lane isolation when a batchCheck sub-batch fails …
SoulPancake Aug 31, 2026
65018d0
Merge remote-tracking branch 'origin/main' into fix/batchcheck-thread…
SoulPancake Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 75 additions & 53 deletions src/main/java/dev/openfga/sdk/api/client/OpenFgaClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -734,6 +735,35 @@ private <T> Stream<List<T>> chunksOf(int chunkSize, List<T> 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 <T> CompletableFuture<Void> executeInLanes(
List<T> items, Function<T, CompletableFuture<Void>> 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<Void> 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
*
Expand Down Expand Up @@ -901,28 +931,20 @@ public CompletableFuture<List<ClientBatchCheckClientResponse>> 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<ClientBatchCheckClientResponse>();

final var clientCheckOptions = options.asClientCheckOptions();

Consumer<ClientCheckRequest> singleClientCheckRequest =
Function<ClientCheckRequest, CompletableFuture<Void>> singleClientCheckRequest =
request -> call(() -> this.check(request, clientCheckOptions))
.handleAsync(ClientBatchCheckClientResponse.asyncHandler(request))
.thenAccept(responses::add)
.thenRun(latch::countDown);
.handle(ClientBatchCheckClientResponse.asyncHandler(request))
.thenAccept(responses::add);
Comment thread
SoulPancake marked this conversation as resolved.

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));
}

/**
Expand Down Expand Up @@ -994,6 +1016,9 @@ public CompletableFuture<ClientBatchCheckResponse> batchCheck(

int maxBatchSize =
options.getMaxBatchSize() != null ? options.getMaxBatchSize() : FgaConstants.CLIENT_MAX_BATCH_SIZE;
if (maxBatchSize <= 0) {
throw new FgaInvalidParameterException("maxBatchSize", "BatchCheck");
}
List<List<BatchCheckItem>> batchedChecks = IntStream.range(
0, (collect.size() + maxBatchSize - 1) / maxBatchSize)
.mapToObj(i -> collect.subList(i * maxBatchSize, Math.min((i + 1) * maxBatchSize, collect.size())))
Expand All @@ -1002,15 +1027,15 @@ public CompletableFuture<ClientBatchCheckResponse> 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");
}
Comment thread
SoulPancake marked this conversation as resolved.
var responses = new ConcurrentLinkedQueue<ClientBatchCheckSingleResponse>();
var failure = new AtomicReference<Throwable>();

var override = new ConfigurationOverride().addHeaders(options);

Consumer<List<BatchCheckItem>> singleBatchCheckRequest = request -> call(() -> {
Function<List<BatchCheckItem>, CompletableFuture<Void>> singleBatchCheckRequest = request -> call(() -> {
BatchCheckRequest body = new BatchCheckRequest().checks(request);
if (options.getConsistency() != null) {
body.consistency(options.getConsistency());
Expand All @@ -1027,42 +1052,39 @@ public CompletableFuture<ClientBatchCheckResponse> batchCheck(

return api.batchCheck(configuration.getStoreId(), body, override);
})
.whenComplete((batchCheckResponseApiResponse, throwable) -> {
try {
if (throwable != null) {
failure.compareAndSet(null, throwable);
return;
}

Map<String, BatchCheckSingleResult> response =
batchCheckResponseApiResponse.getData().getResult();

List<ClientBatchCheckSingleResponse> 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();
.<Void>handle((batchCheckResponseApiResponse, throwable) -> {
if (throwable != null) {
failure.compareAndSet(null, throwable);
return null;
}

Map<String, BatchCheckSingleResult> response =
batchCheckResponseApiResponse.getData().getResult();
Comment thread
SoulPancake marked this conversation as resolved.

List<ClientBatchCheckSingleResponse> 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;
});
Comment thread
SoulPancake marked this conversation as resolved.

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)));
});
}

/**
Expand Down
Loading
Loading