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 832f3601d3ea CAMEL-26826: camel-opa - bound response body consumption
by request timeout (#29628)
832f3601d3ea is described below
commit 832f3601d3eaa4737a1c1bf186a8d801f1efdd0b
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 12 17:07:45 2026 +0100
CAMEL-26826: camel-opa - bound response body consumption by request timeout
(#29628)
Apply requestTimeout to the complete OPA REST response, including body
consumption, and cancel unfinished exchanges on timeout or interruption.
Preserve fatal Error causes. Includes stalled-body timeout-cause assertions,
finite truncated-JSON regressions for both fail-open settings, and transport
documentation.
Co-authored-by: Codex <[email protected]>
Co-Authored-By: Claude Fable 5.1 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../apache/camel/catalog/docs/opa-component.adoc | 5 +-
.../camel-opa/src/main/docs/opa-component.adoc | 5 +-
.../apache/camel/component/opa/OpaHttpClient.java | 44 ++++++-
.../camel/component/opa/OpaRestTransportTest.java | 142 ++++++++++++++++++++-
4 files changed, 188 insertions(+), 8 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opa-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opa-component.adoc
index 70c72ed5b535..f05907d8b19a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opa-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/opa-component.adoc
@@ -487,7 +487,10 @@ separate process that a timeout can abandon.
In `rest` mode the component talks to OPA over HTTP, and both waits are
bounded:
* `connectionTimeout` (default 10s) — establishing the connection.
-* `requestTimeout` (default 30s) — waiting for the decision once connected.
+* `requestTimeout` (default 30s) — receiving the complete response for each
HTTP request, including its body.
+
+Receiving response headers does not end the request deadline: a server that
sends headers and then stalls while
+writing the decision still times out. SDK retries, when applicable, make
separate HTTP requests with the same limit.
Neither is optional in practice. A refused connection fails immediately, but a
server that *accepts* and then
stops answering — wedged, mid-restart, or behind a load balancer holding the
socket — would otherwise park the
diff --git a/components/camel-opa/src/main/docs/opa-component.adoc
b/components/camel-opa/src/main/docs/opa-component.adoc
index 70c72ed5b535..f05907d8b19a 100644
--- a/components/camel-opa/src/main/docs/opa-component.adoc
+++ b/components/camel-opa/src/main/docs/opa-component.adoc
@@ -487,7 +487,10 @@ separate process that a timeout can abandon.
In `rest` mode the component talks to OPA over HTTP, and both waits are
bounded:
* `connectionTimeout` (default 10s) — establishing the connection.
-* `requestTimeout` (default 30s) — waiting for the decision once connected.
+* `requestTimeout` (default 30s) — receiving the complete response for each
HTTP request, including its body.
+
+Receiving response headers does not end the request deadline: a server that
sends headers and then stalls while
+writing the decision still times out. SDK retries, when applicable, make
separate HTTP requests with the same limit.
Neither is optional in practice. A refused connection fails immediately, but a
server that *accepts* and then
stops answering — wedged, mid-restart, or behind a load balancer holding the
socket — would otherwise park the
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHttpClient.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHttpClient.java
index bd537e6173b9..2694ae1264ad 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHttpClient.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHttpClient.java
@@ -16,12 +16,18 @@
*/
package org.apache.camel.component.opa;
+import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
+import java.net.http.HttpTimeoutException;
import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import javax.net.ssl.SSLContext;
@@ -38,11 +44,13 @@ import org.apache.camel.util.ObjectHelper;
* are the JDK default of "wait indefinitely". A server that accepts the
connection and then goes quiet parks the
* calling thread for ever - and a component that fails closed never reaches
the point of denying, it simply stops.
* {@code failOpen} does not help, because it sits downstream of a call that
never returns.</li>
- * <li><b>It builds a client per request.</b> On the Java 17 baseline {@link
HttpClient} is not {@link AutoCloseable},
- * so each one holds its selector thread and executor until it is collected -
once per message, on the path an
- * {@code OpaSecurityPolicy} guards.</li>
+ * <li><b>It builds a client per request.</b> Each {@link HttpClient} holds
its selector thread and executor until it is
+ * closed or collected - once per message, on the path an {@code
OpaSecurityPolicy} guards.</li>
* </ul>
- * One client is built here per evaluator and reused, and every request is
re-issued carrying a timeout.
+ * One client is built here per evaluator and reused, and every request is
re-issued carrying a timeout. The response
+ * body is buffered rather than streamed with {@code ofInputStream}: the
timeout of a request only covers the wait for
+ * the response headers, so the whole response, body included, is awaited
under {@code requestTimeout} here, and a
+ * server that stalls or drips its body is cut off like one that never answers.
*/
public class OpaHttpClient implements HTTPClient, AutoCloseable {
@@ -85,6 +93,32 @@ public class OpaHttpClient implements HTTPClient,
AutoCloseable {
if (bearerToken != null) {
builder.header(AUTHORIZATION, "Bearer " + bearerToken);
}
- return client.send(builder.build(),
HttpResponse.BodyHandlers.ofInputStream());
+ // A streaming body handler completes as soon as headers arrive,
leaving the SDK's subsequent body read
+ // outside HttpRequest.timeout(). Complete the body before returning
its stream and bound that entire wait.
+ CompletableFuture<HttpResponse<InputStream>> response =
client.sendAsync(builder.build(),
+ info ->
HttpResponse.BodySubscribers.mapping(HttpResponse.BodySubscribers.ofByteArray(),
+ bytes -> new ByteArrayInputStream(bytes)));
+ try {
+ return response.get(requestTimeout.toMillis(),
TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ response.cancel(true);
+ throw new HttpTimeoutException("Timed out waiting for the complete
OPA response after " + requestTimeout);
+ } catch (InterruptedException e) {
+ response.cancel(true);
+ throw e;
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause();
+ if (cause instanceof Error error) {
+ // an OutOfMemoryError from buffering the body is not an HTTP
failure; keep it what it is
+ throw error;
+ }
+ if (cause instanceof IOException io) {
+ throw io;
+ }
+ if (cause instanceof RuntimeException runtime) {
+ throw runtime;
+ }
+ throw new IOException("OPA HTTP request failed", cause);
+ }
}
}
diff --git
a/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaRestTransportTest.java
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaRestTransportTest.java
index 8bb3096a273f..6c70565f1ca0 100644
---
a/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaRestTransportTest.java
+++
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaRestTransportTest.java
@@ -21,10 +21,17 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
+import java.net.URI;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.net.http.HttpTimeoutException;
import java.nio.charset.StandardCharsets;
import java.security.KeyStore;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ssl.KeyManagerFactory;
@@ -41,6 +48,8 @@ import org.apache.camel.test.junit6.CamelTestSupport;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import static org.assertj.core.api.Assertions.assertThat;
@@ -182,6 +191,136 @@ public class OpaRestTransportTest extends
CamelTestSupport {
return server.getAddress().getPort();
}
+ /** Sends response headers and an incomplete JSON document, then waits for
test cleanup. */
+ private int stalledBodyPort(CountDownLatch headersSent) throws IOException
{
+ CountDownLatch release = new CountDownLatch(1);
+ HttpServer server = HttpServer.create(new
InetSocketAddress("localhost", 0), 0);
+ server.createContext("/v1/data/authz/allow", exchange -> {
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, 0);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write('{');
+ out.flush();
+ headersSent.countDown();
+ try {
+ release.await(60, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ });
+ server.start();
+ open.add(() -> {
+ release.countDown();
+ server.stop(0);
+ });
+ return server.getAddress().getPort();
+ }
+
+ @Test
+ @Timeout(10)
+ void failsClosedWhenTheResponseBodyStallsAfterHeaders() throws Exception {
+ CountDownLatch headersSent = new CountDownLatch(1);
+ Exchange out = template.request(
+ "opa:authz/allow?serverUrl=http://localhost:"; +
stalledBodyPort(headersSent)
+ + "&requestTimeout=500",
+ e -> {
+ });
+
+ // the timeout, not some other failure, is what stopped the evaluation
+
assertThat(out.getException()).isInstanceOf(OpaPolicyEvaluationException.class)
+ .hasRootCauseInstanceOf(HttpTimeoutException.class);
+
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_ALLOW)).isNull();
+ assertThat(headersSent.getCount()).isZero();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = { false, true })
+ @Timeout(10)
+ void failsClosedWhenTheServerClosesAfterTruncatedJson(boolean failOpen)
throws Exception {
+ HttpServer server = HttpServer.create(new
InetSocketAddress("localhost", 0), 0);
+ server.createContext("/v1/data/authz/allow", exchange -> {
+ exchange.getResponseHeaders().add("Content-Type",
"application/json");
+ exchange.sendResponseHeaders(200, 1);
+ try (OutputStream body = exchange.getResponseBody()) {
+ body.write('{');
+ }
+ });
+ server.start();
+ open.add(() -> server.stop(0));
+
+ Exchange out = template.request(
+ "opa:authz/allow?serverUrl=http://localhost:"; +
server.getAddress().getPort()
+ + "&requestTimeout=1000&failOpen=" +
failOpen,
+ e -> {
+ });
+
+
assertThat(out.getException()).isInstanceOf(OpaPolicyEvaluationException.class);
+
assertThat(out.getException()).hasRootCauseInstanceOf(com.fasterxml.jackson.core.JsonParseException.class);
+
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_ALLOW)).isNull();
+
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_FAILED_OPEN)).isNull();
+ }
+
+ @Test
+ @Timeout(10)
+ void reportsFailOpenWhenTheResponseBodyStallsAfterHeaders() throws
Exception {
+ CountDownLatch headersSent = new CountDownLatch(1);
+ Exchange out = template.request(
+ "opa:authz/allow?serverUrl=http://localhost:"; +
stalledBodyPort(headersSent)
+ + "&requestTimeout=500&failOpen=true",
+ e -> {
+ });
+
+ assertThat(out.getException()).isNull();
+
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_ALLOW)).isEqualTo(true);
+
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_FAILED_OPEN)).isEqualTo(true);
+ assertThat(headersSent.getCount()).isZero();
+ }
+
+ @Test
+ @Timeout(10)
+ void interruptsTheWaitForAStalledResponseBody() throws Exception {
+ CountDownLatch headersSent = new CountDownLatch(1);
+ CountDownLatch interrupted = new CountDownLatch(1);
+ HttpRequest request = HttpRequest.newBuilder(
+ URI.create("http://localhost:"; + stalledBodyPort(headersSent)
+ "/v1/data/authz/allow")).build();
+
+ try (OpaHttpClient transport = new OpaHttpClient(1000, 30000, null,
null);
+ var executor = Executors.newSingleThreadExecutor()) {
+ var call = executor.submit(() -> {
+ try {
+ transport.send(request);
+ } catch (InterruptedException e) {
+ interrupted.countDown();
+ Thread.currentThread().interrupt();
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ });
+ assertThat(headersSent.await(5, TimeUnit.SECONDS)).isTrue();
+ assertThat(call.cancel(true)).isTrue();
+ assertThat(interrupted.await(5, TimeUnit.SECONDS)).isTrue();
+ }
+ }
+
+ @Test
+ @Timeout(10)
+ void retainsResponseMetadataAndTheCompletedBody() throws Exception {
+ URI uri = URI.create("http://localhost:"; + recordingPort(new
AtomicReference<>()) + "/v1/data/authz/allow");
+ HttpRequest request = HttpRequest.newBuilder(uri).build();
+
+ try (OpaHttpClient transport = new OpaHttpClient(1000, 5000, null,
null)) {
+ HttpResponse<InputStream> response = transport.send(request);
+ assertThat(response.statusCode()).isEqualTo(200);
+ assertThat(response.uri()).isEqualTo(uri);
+ assertThat(response.request().uri()).isEqualTo(request.uri());
+
assertThat(response.headers().firstValue("Content-Type")).contains("application/json");
+ try (InputStream body = response.body()) {
+ assertThat(new String(body.readAllBytes(),
StandardCharsets.UTF_8)).isEqualTo("{\"result\":true}");
+ }
+ }
+ }
+
@Test
@Timeout(60)
void sendsTheBearerTokenWhenOneIsConfigured() throws Exception {
@@ -215,7 +354,8 @@ public class OpaRestTransportTest extends CamelTestSupport {
"opa:authz/allow?serverUrl=http://localhost:"; + mutePort() +
"&requestTimeout=500", e -> {
});
-
assertThat(out.getException()).isInstanceOf(OpaPolicyEvaluationException.class);
+
assertThat(out.getException()).isInstanceOf(OpaPolicyEvaluationException.class)
+ .hasRootCauseInstanceOf(HttpTimeoutException.class);
assertThat(out.getMessage().getHeader(OpaConstants.DECISION_ALLOW)).isNull();
}
