This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 411e25ec05f7 CAMEL-26408: camel-openfga - scale the batchCheck await
timeout by the fan-out (#28703)
411e25ec05f7 is described below
commit 411e25ec05f7fb476e4feefe7f01ea30f41af29f
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Oct 29 22:38:00 2026 +0100
CAMEL-26408: camel-openfga - scale the batchCheck await timeout by the
fan-out (#28703)
batchCheck issues one check per object, maxParallelRequests at a time, so
ceil(n / maxParallelRequests) waves share a single future. It was bounded by
the single-request await timeout (connect+read, per attempt, across the
retries,
plus a margin), so a large batch against a slow server could be cancelled
mid-flight and surfaced as a spurious "Timed out after ...". Scale the
bound by
the number of waves - the safety margin still added only once - and report
the
actual bound in the timeout message.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/component/openfga/OpenFgaAuthorizer.java | 39 +++++++++++---
.../camel/component/openfga/OpenFgaProducer.java | 8 ++-
.../component/openfga/OpenFgaAuthorizerTest.java | 59 ++++++++++++++++++++++
3 files changed, 97 insertions(+), 9 deletions(-)
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaAuthorizer.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaAuthorizer.java
index 25a97ee59c55..065354464831 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaAuthorizer.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaAuthorizer.java
@@ -74,6 +74,7 @@ public class OpenFgaAuthorizer {
private final Expression continuationToken;
private final ConsistencyPreference consistency;
private final boolean failOpen;
+ private final long perRequestTimeoutMillis;
private final long awaitTimeoutMillis;
public OpenFgaAuthorizer(OpenFgaClient client, OpenFgaConfiguration
configuration, CamelContext camelContext) {
@@ -88,7 +89,8 @@ public class OpenFgaAuthorizer {
this.continuationToken = compile(camelContext,
configuration.getContinuationToken());
this.consistency = parseConsistency(configuration.getConsistency());
this.failOpen = configuration.isFailOpen();
- this.awaitTimeoutMillis = awaitTimeout(configuration);
+ this.perRequestTimeoutMillis = perRequestTimeout(configuration);
+ this.awaitTimeoutMillis = perRequestTimeoutMillis +
AWAIT_SAFETY_MARGIN_MILLIS;
}
/**
@@ -157,13 +159,27 @@ public class OpenFgaAuthorizer {
}
/**
- * The longest one exchange may wait for an answer: what the SDK itself
allows for - a connect and a request per
- * attempt, across the first attempt and every retry - plus a margin.
+ * What the SDK itself allows for one request - a connect and a request
per attempt, across the first attempt and
+ * every retry. The safety margin is not included here: {@link
#awaitTimeoutMillis} adds it once for a single call,
+ * and {@link #batchAwaitTimeoutMillis} adds it once on top of the whole
fan-out.
*/
- private static long awaitTimeout(OpenFgaConfiguration configuration) {
+ private static long perRequestTimeout(OpenFgaConfiguration configuration) {
long perAttempt = configuration.getConnectTimeout() +
configuration.getReadTimeout();
long attempts = Math.max(0, configuration.getMaxRetries()) + 1L;
- return perAttempt * attempts + AWAIT_SAFETY_MARGIN_MILLIS;
+ return perAttempt * attempts;
+ }
+
+ /**
+ * The longest a {@code batchCheck} of {@code objectCount} objects may
wait. The SDK issues the batch as one request
+ * per object with at most {@code maxParallelRequests} in flight, so the
fan-out runs in
+ * {@code ceil(objectCount / maxParallelRequests)} serialized waves.
Bounding the whole batch by the single-request
+ * timeout would cancel a large batch against a slow server mid-flight and
report it as a spurious timeout, so the
+ * per-request time is multiplied by the number of waves - the safety
margin is still added only once.
+ */
+ long batchAwaitTimeoutMillis(int objectCount, int maxParallelRequests) {
+ int parallel = Math.max(1, maxParallelRequests);
+ long waves = Math.max(1L, (objectCount + parallel - 1L) / parallel);
+ return perRequestTimeoutMillis * waves + AWAIT_SAFETY_MARGIN_MILLIS;
}
/**
@@ -442,6 +458,15 @@ public class OpenFgaAuthorizer {
* routing thread for ever.
*/
<T> T await(Exchange exchange, String what, CallSupplier<T> call) throws
OpenFgaEvaluationException {
+ return await(exchange, what, call, awaitTimeoutMillis);
+ }
+
+ /**
+ * As {@link #await(Exchange, String, CallSupplier)} but with an explicit
bound, for the batch path whose fan-out
+ * needs more than a single request's budget (see {@link
#batchAwaitTimeoutMillis}).
+ */
+ <T> T await(Exchange exchange, String what, CallSupplier<T> call, long
timeoutMillis)
+ throws OpenFgaEvaluationException {
CompletableFuture<T> future;
try {
future = call.get();
@@ -449,7 +474,7 @@ public class OpenFgaAuthorizer {
throw new OpenFgaEvaluationException("Could not build the OpenFGA
request to " + what, exchange, e);
}
try {
- return future.get(awaitTimeoutMillis, TimeUnit.MILLISECONDS);
+ return future.get(timeoutMillis, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
// a shutdown, not a failure of the decision point. Restore the
flag the interruptible wait cleared and
// fail: nothing decided that this exchange was permitted, so
failOpen must not claim it did either.
@@ -461,7 +486,7 @@ public class OpenFgaAuthorizer {
} catch (TimeoutException e) {
future.cancel(true);
throw new OpenFgaEvaluationException(
- "Timed out after " + awaitTimeoutMillis + "ms waiting for
OpenFGA to " + what, exchange, e);
+ "Timed out after " + timeoutMillis + "ms waiting for
OpenFGA to " + what, exchange, e);
} catch (ExecutionException e) {
// the SDK reports an unreachable server, an HTTP error and a
serialization problem all through the
// future; every one of them is the same thing to a route - no
verdict - so they are reported alike
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducer.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducer.java
index 1972ecb5f068..a33c8d511799 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducer.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducer.java
@@ -180,8 +180,9 @@ public class OpenFgaProducer extends DefaultProducer {
return;
}
+ int maxParallelRequests =
getEndpoint().getConfiguration().getMaxParallelRequests();
ClientBatchCheckClientOptions options = new
ClientBatchCheckClientOptions()
-
.maxParallelRequests(getEndpoint().getConfiguration().getMaxParallelRequests());
+ .maxParallelRequests(maxParallelRequests);
if
(ObjectHelper.isNotEmpty(getEndpoint().getConfiguration().getAuthorizationModelId()))
{
options.authorizationModelId(getEndpoint().getConfiguration().getAuthorizationModelId());
}
@@ -191,7 +192,10 @@ public class OpenFgaProducer extends DefaultProducer {
List<ClientBatchCheckClientResponse> responses = authorizer.await(
exchange, "batch check " + relation + " for " + user,
- () -> authorizer.getClient().clientBatchCheck(requests,
options));
+ () -> authorizer.getClient().clientBatchCheck(requests,
options),
+ // the batch fans out one request per object,
maxParallelRequests at a time, so it needs more than a
+ // single request's budget - otherwise a large batch against a
slow server is cancelled mid-flight
+ authorizer.batchAwaitTimeoutMillis(requests.size(),
maxParallelRequests));
Set<String> permitted = new HashSet<>();
for (ClientBatchCheckClientResponse response : responses) {
diff --git
a/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaAuthorizerTest.java
b/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaAuthorizerTest.java
new file mode 100644
index 000000000000..eb1595db1d0f
--- /dev/null
+++
b/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaAuthorizerTest.java
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.openfga;
+
+import dev.openfga.sdk.api.client.OpenFgaClient;
+import org.apache.camel.CamelContext;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * Unit tests for how {@link OpenFgaAuthorizer} bounds the wait on a request.
+ */
+class OpenFgaAuthorizerTest {
+
+ private final OpenFgaClient client = mock(OpenFgaClient.class);
+ private final CamelContext camelContext = mock(CamelContext.class);
+
+ @Test
+ void batchAwaitTimeoutScalesWithTheNumberOfFanOutWaves() {
+ OpenFgaConfiguration configuration = new OpenFgaConfiguration();
+ configuration.setConnectTimeout(10000);
+ configuration.setReadTimeout(10000);
+ configuration.setMaxRetries(3);
+ OpenFgaAuthorizer authorizer = new OpenFgaAuthorizer(client,
configuration, camelContext);
+
+ // connect + read, per attempt, across the first try and the three
retries
+ long perRequest = (10000L + 10000L) * (3 + 1);
+ // AWAIT_SAFETY_MARGIN_MILLIS, added once however many waves the batch
fans out into
+ long margin = 5000L;
+
+ // a batch that fits in a single wave is bounded like a single
request: the per-request time plus the margin
+ assertThat(authorizer.batchAwaitTimeoutMillis(10,
10)).isEqualTo(perRequest + margin);
+ assertThat(authorizer.batchAwaitTimeoutMillis(5,
10)).isEqualTo(perRequest + margin);
+ // more objects than can run in parallel fan out into ceil(objectCount
/ maxParallelRequests) waves, and the
+ // bound grows a whole per-request budget with each extra wave - while
the margin is still added only once
+ assertThat(authorizer.batchAwaitTimeoutMillis(11,
10)).isEqualTo(perRequest * 2 + margin);
+ assertThat(authorizer.batchAwaitTimeoutMillis(25,
10)).isEqualTo(perRequest * 3 + margin);
+ // defensive: a non-positive parallelism is clamped to one in-flight
request rather than dividing by zero
+ assertThat(authorizer.batchAwaitTimeoutMillis(5,
0)).isEqualTo(perRequest * 5 + margin);
+ // an empty batch still bounds a single wave rather than collapsing to
just the margin
+ assertThat(authorizer.batchAwaitTimeoutMillis(0,
10)).isEqualTo(perRequest + margin);
+ }
+}
