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 188ed5bdad79 CAMEL-26827: release OPA and OpenFGA health probe clients
on stop (#29630)
188ed5bdad79 is described below
commit 188ed5bdad7997213de7542eb9cc04fc97f6bdee
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 12 11:22:36 2026 +0100
CAMEL-26827: release OPA and OpenFGA health probe clients on stop (#29630)
* CAMEL-26827: release OPA and OpenFGA health probe clients on stop
Own probe transports per health check instead of retaining TLS clients in
static maps. Close them when producers or the last guarded route stop,
retaining compatible static probe entry points and independent restart behavior.
Co-authored-by: Codex <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
* CAMEL-26827: Address review feedback - the probe call is bounded as a
whole, and the tests read the client through an accessor
The timeout of an HttpRequest only covers the wait for the response headers,
so a health endpoint that stalled while sending its body held the thread
that
runs the check. Both probes now send asynchronously and wait for the whole
response under the probe timeout, cancelling the call on timeout or
interrupt.
The lifecycle tests read the probe's client through a package-private
accessor instead of reflection on its field.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---------
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Codex <[email protected]>
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../apache/camel/component/opa/OpaHealthProbe.java | 88 ++++++++++----
.../apache/camel/component/opa/OpaProducer.java | 1 +
.../component/opa/OpaProducerHealthCheck.java | 13 ++-
.../component/opa/security/OpaSecurityPolicy.java | 1 +
.../opa/security/OpaSecurityPolicyHealthCheck.java | 13 ++-
.../component/opa/OpaHealthProbeLifecycleTest.java | 127 ++++++++++++++++++++
.../component/openfga/OpenFgaHealthProbe.java | 89 +++++++++-----
.../camel/component/openfga/OpenFgaProducer.java | 1 +
.../openfga/OpenFgaProducerHealthCheck.java | 13 ++-
.../openfga/security/OpenFgaSecurityPolicy.java | 1 +
.../security/OpenFgaSecurityPolicyHealthCheck.java | 13 ++-
.../openfga/OpenFgaHealthProbeLifecycleTest.java | 128 +++++++++++++++++++++
12 files changed, 421 insertions(+), 67 deletions(-)
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHealthProbe.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHealthProbe.java
index 1f9daf4ebc53..c1ef4af7618f 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHealthProbe.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaHealthProbe.java
@@ -20,9 +20,12 @@ import java.net.URI;
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.Map;
-import java.util.concurrent.ConcurrentHashMap;
+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;
@@ -37,37 +40,66 @@ import org.apache.camel.util.URISupport;
* Both ask the same question of the same endpoint, so the probe lives in one
place: a fix here - a changed timeout, a
* new failure mode to report - applies to both rather than to whichever was
remembered.
*/
-public final class OpaHealthProbe {
+public final class OpaHealthProbe implements AutoCloseable {
private static final Duration TIMEOUT = Duration.ofSeconds(5);
- /**
- * Shared across every OPA health check in the JVM. {@link HttpClient}
only became {@link AutoCloseable} in Java 21,
- * so on the Java 17 baseline one client per check would leak its selector
thread with no way to shut it down. The
- * client is immutable and thread-safe, so sharing is free; the
per-request URL and token live on the
- * {@link HttpRequest}.
- */
- private static final HttpClient HTTP_CLIENT =
HttpClient.newBuilder().connectTimeout(TIMEOUT).build();
+ private final SSLContext sslContext;
+ private HttpClient client;
+ private boolean closed;
+
+ public OpaHealthProbe(SSLContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ private synchronized HttpClient client() {
+ if (closed) {
+ throw new IllegalStateException("The health probe is closed");
+ }
+ if (client == null) {
+ HttpClient.Builder builder =
HttpClient.newBuilder().connectTimeout(TIMEOUT);
+ if (sslContext != null) {
+ builder.sslContext(sslContext);
+ }
+ client = builder.build();
+ }
+ return client;
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ if (client != null) {
+ client.shutdownNow();
+ client = null;
+ }
+ }
/**
- * One client per distinct TLS configuration, because an {@link
SSLContext} can only be set when the client is
- * built. A probe that ignored it would fail its handshake against the
very server the decision call reaches happily
- * - reporting DOWN, and with it an application that never becomes ready.
- * <p/>
- * Keyed on identity and never evicted, which is bounded in practice: the
key is an {@code SSLContext} built from an
- * endpoint's {@code sslContextParameters}, and a deployment has one or
two of those, not one per exchange.
+ * The client this probe holds, without creating one. For the tests of the
probe's lifecycle.
*/
- private static final Map<SSLContext, HttpClient> TLS_CLIENTS = new
ConcurrentHashMap<>();
-
- private OpaHealthProbe() {
+ synchronized HttpClient currentClient() {
+ return client;
}
- private static HttpClient clientFor(SSLContext sslContext) {
- if (sslContext == null) {
- return HTTP_CLIENT;
+ /**
+ * Sends the request and waits for the whole response, body included, for
at most {@link #TIMEOUT}. The timeout of
+ * the request itself only covers the wait for the response headers, so a
server that stalls while sending the body
+ * would otherwise hold the thread that runs the health check for as long
as it likes.
+ */
+ private <T> HttpResponse<T> sendBounded(HttpRequest request,
HttpResponse.BodyHandler<T> handler) throws Exception {
+ CompletableFuture<HttpResponse<T>> call = client().sendAsync(request,
handler);
+ try {
+ return call.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ call.cancel(true);
+ throw new HttpTimeoutException("Timed out waiting for the OPA
health endpoint after " + TIMEOUT);
+ } catch (InterruptedException e) {
+ call.cancel(true);
+ throw e;
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof Exception cause ? cause : e;
}
- return TLS_CLIENTS.computeIfAbsent(sslContext,
- ctx ->
HttpClient.newBuilder().connectTimeout(TIMEOUT).sslContext(ctx).build());
}
/**
@@ -85,6 +117,12 @@ public final class OpaHealthProbe {
public static void probe(
HealthCheckResultBuilder builder, String serverUrl, String
bearerToken, String policyPath,
SSLContext sslContext) {
+ try (OpaHealthProbe probe = new OpaHealthProbe(sslContext)) {
+ probe.probe(builder, serverUrl, bearerToken, policyPath);
+ }
+ }
+
+ public void probe(HealthCheckResultBuilder builder, String serverUrl,
String bearerToken, String policyPath) {
builder.detail("opa.serverUrl", URISupport.sanitizeUri(serverUrl));
builder.detail("opa.policyPath", policyPath);
@@ -99,7 +137,7 @@ public final class OpaHealthProbe {
}
try {
- HttpResponse<Void> response =
clientFor(sslContext).send(request.build(),
HttpResponse.BodyHandlers.discarding());
+ HttpResponse<Void> response = sendBounded(request.build(),
HttpResponse.BodyHandlers.discarding());
if (response.statusCode() == 200) {
builder.up();
} else {
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducer.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducer.java
index 0dbaa901a1ef..d571fe334b73 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducer.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducer.java
@@ -71,6 +71,7 @@ public class OpaProducer extends DefaultProducer {
protected void doStop() throws Exception {
if (healthCheckRepository != null && producerHealthCheck != null) {
healthCheckRepository.removeHealthCheck(producerHealthCheck);
+ producerHealthCheck.close();
producerHealthCheck = null;
}
super.doStop();
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducerHealthCheck.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducerHealthCheck.java
index 6305a4de280c..91385f25bfc5 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducerHealthCheck.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaProducerHealthCheck.java
@@ -31,12 +31,12 @@ import org.apache.camel.util.URISupport;
* check probes the server's {@code /health} endpoint so that an unavailable
policy decision point is visible before
* traffic starts failing, rather than only in the error logs afterwards.
*/
-public class OpaProducerHealthCheck extends AbstractHealthCheck {
+public class OpaProducerHealthCheck extends AbstractHealthCheck implements
AutoCloseable {
private final String serverUrl;
private final String bearerToken;
private final String policyPath;
- private final SSLContext sslContext;
+ private final OpaHealthProbe probe;
public OpaProducerHealthCheck(String serverUrl, String bearerToken, String
policyPath, String id,
SSLContext sslContext) {
@@ -46,11 +46,16 @@ public class OpaProducerHealthCheck extends
AbstractHealthCheck {
this.serverUrl = serverUrl;
this.bearerToken = bearerToken;
this.policyPath = policyPath;
- this.sslContext = sslContext;
+ this.probe = new OpaHealthProbe(sslContext);
}
@Override
protected void doCall(HealthCheckResultBuilder builder, Map<String,
Object> options) {
- OpaHealthProbe.probe(builder, serverUrl, bearerToken, policyPath,
sslContext);
+ probe.probe(builder, serverUrl, bearerToken, policyPath);
+ }
+
+ @Override
+ public void close() {
+ probe.close();
}
}
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicy.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicy.java
index af4f7bc11a71..3dd52eaabdc2 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicy.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicy.java
@@ -274,6 +274,7 @@ public class OpaSecurityPolicy implements
AuthorizationPolicy {
if (registry != null) {
registry.unregister(healthCheck);
}
+ healthCheck.close();
healthCheck = null;
}
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicyHealthCheck.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicyHealthCheck.java
index 2c53a3447005..de154182c1c0 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicyHealthCheck.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/security/OpaSecurityPolicyHealthCheck.java
@@ -33,12 +33,12 @@ import org.apache.camel.util.URISupport;
* unreachable server here fails every message outright, which is exactly the
condition worth surfacing before traffic
* arrives rather than after.
*/
-public class OpaSecurityPolicyHealthCheck extends AbstractHealthCheck {
+public class OpaSecurityPolicyHealthCheck extends AbstractHealthCheck
implements AutoCloseable {
private final String serverUrl;
private final String bearerToken;
private final String policyPath;
- private final SSLContext sslContext;
+ private final OpaHealthProbe probe;
public OpaSecurityPolicyHealthCheck(String serverUrl, String bearerToken,
String policyPath,
SSLContext sslContext) {
@@ -48,11 +48,16 @@ public class OpaSecurityPolicyHealthCheck extends
AbstractHealthCheck {
this.serverUrl = serverUrl;
this.bearerToken = bearerToken;
this.policyPath = policyPath;
- this.sslContext = sslContext;
+ this.probe = new OpaHealthProbe(sslContext);
}
@Override
protected void doCall(HealthCheckResultBuilder builder, Map<String,
Object> options) {
- OpaHealthProbe.probe(builder, serverUrl, bearerToken, policyPath,
sslContext);
+ probe.probe(builder, serverUrl, bearerToken, policyPath);
+ }
+
+ @Override
+ public void close() {
+ probe.close();
}
}
diff --git
a/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaHealthProbeLifecycleTest.java
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaHealthProbeLifecycleTest.java
new file mode 100644
index 000000000000..aeae0129681c
--- /dev/null
+++
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaHealthProbeLifecycleTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.opa;
+
+import java.lang.reflect.Field;
+import java.net.http.HttpClient;
+import java.time.Duration;
+import java.util.List;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.camel.Producer;
+import org.apache.camel.component.opa.security.OpaSecurityPolicyHealthCheck;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.WritableHealthCheckRepository;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.jsse.SSLContextParameters;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class OpaHealthProbeLifecycleTest {
+
+ @Test
+ void closesProducerAndPolicyProbeClientsIndependently() throws Exception {
+ SSLContext ssl = SSLContext.getDefault();
+ try (OpaProducerHealthCheck producer = new OpaProducerHealthCheck(
+ "http://localhost:1";, null, "scope", "endpoint", ssl);
+ OpaSecurityPolicyHealthCheck policy = new
OpaSecurityPolicyHealthCheck(
+ "http://localhost:1";, null, "scope", ssl)) {
+ for (HealthCheck check : List.of(producer, policy)) {
+ check.setEnabled(true);
+
assertThat(check.call().getState()).isEqualTo(HealthCheck.State.DOWN);
+ }
+ HttpClient producerClient = client(producer);
+ HttpClient policyClient = client(policy);
+ producer.close();
+
+
await().atMost(Duration.ofSeconds(5)).until(producerClient::isTerminated);
+ assertThat(policyClient.isTerminated()).isFalse();
+ assertThat(producer.call().getError()).isPresent()
+ .get().isInstanceOf(IllegalStateException.class);
+ policy.close();
+
await().atMost(Duration.ofSeconds(5)).until(policyClient::isTerminated);
+ }
+ }
+
+ @Test
+ void releasesEachNewTlsConfigurationOnClose() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ SSLContext ssl = SSLContext.getInstance("TLS");
+ ssl.init(null, null, null);
+ try (OpaProducerHealthCheck check = new OpaProducerHealthCheck(
+ "http://localhost:1";, null, "scope", "endpoint", ssl)) {
+ check.setEnabled(true);
+ check.call();
+ HttpClient client = client(check);
+ check.close();
+ check.close();
+
await().atMost(Duration.ofSeconds(5)).until(client::isTerminated);
+
assertThat(check.call().getState()).isEqualTo(HealthCheck.State.DOWN);
+ assertThat(check.call().getError()).isPresent()
+ .get().isInstanceOf(IllegalStateException.class);
+ }
+ }
+ }
+
+ @Test
+ void producerRestartClosesTheOldClientAndCreatesAnIndependentProbe()
throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ context.getComponent("opa",
OpaComponent.class).setHealthCheckProducerEnabled(true);
+ context.getRegistry().bind("probeTls", new SSLContextParameters());
+ context.start();
+ Producer producer = context.getEndpoint(
+
"opa:authz/allow?serverUrl=http://localhost:1&sslContextParameters=#probeTls";).createProducer();
+ context.addService(producer, true);
+ try {
+ WritableHealthCheckRepository repository =
HealthCheckHelper.getHealthCheckRepository(
+ context, "producers",
WritableHealthCheckRepository.class);
+ assertThat(repository).isNotNull();
+ repository.setEnabled(true);
+ HealthCheck first = repository.stream()
+ .filter(check ->
check.getId().startsWith("producer:opa-")).findFirst().orElseThrow();
+ first.call();
+ HttpClient firstClient = client(first);
+ producer.stop();
+
await().atMost(Duration.ofSeconds(5)).until(firstClient::isTerminated);
+ assertThat(repository.stream().filter(check ->
check.getId().startsWith("producer:opa-"))).isEmpty();
+
+ producer.start();
+ HealthCheck restarted = repository.stream()
+ .filter(check ->
check.getId().startsWith("producer:opa-")).findFirst().orElseThrow();
+ assertThat(restarted).isNotSameAs(first);
+ restarted.call();
+ HttpClient restartedClient = client(restarted);
+ assertThat(restartedClient).isNotSameAs(firstClient);
+ context.stop();
+
await().atMost(Duration.ofSeconds(5)).until(restartedClient::isTerminated);
+ } finally {
+ producer.stop();
+ }
+ }
+ }
+
+ private static HttpClient client(HealthCheck check) throws Exception {
+ Field probe = check.getClass().getDeclaredField("probe");
+ probe.setAccessible(true);
+ // the probe exposes its client to this package, so a rename there is
a compile error, not a surprise
+ return ((OpaHealthProbe) probe.get(check)).currentClient();
+ }
+}
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaHealthProbe.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaHealthProbe.java
index d57fa1a4bb39..dd6b5e644272 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaHealthProbe.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaHealthProbe.java
@@ -20,9 +20,12 @@ import java.net.URI;
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.Map;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.regex.Pattern;
import javax.net.ssl.SSLContext;
@@ -38,7 +41,7 @@ import org.apache.camel.util.URISupport;
* Both ask the same question of the same server, so the probe lives in one
place: a fix here - a changed timeout, a new
* failure mode to report - applies to both rather than to whichever was
remembered.
*/
-public final class OpenFgaHealthProbe {
+public final class OpenFgaHealthProbe implements AutoCloseable {
private static final Duration TIMEOUT = Duration.ofSeconds(5);
@@ -52,33 +55,62 @@ public final class OpenFgaHealthProbe {
*/
private static final Pattern SERVING =
Pattern.compile("\"status\"\\s*:\\s*\"SERVING\"");
- /**
- * Shared across every OpenFGA health check in the JVM. {@link HttpClient}
only became {@link AutoCloseable} in Java
- * 21, so on the Java 17 baseline one client per check would leak its
selector thread with no way to shut it down.
- * The client is immutable and thread-safe, so sharing is free; the
per-request URL and token live on the
- * {@link HttpRequest}.
- */
- private static final HttpClient HTTP_CLIENT =
HttpClient.newBuilder().connectTimeout(TIMEOUT).build();
+ private final SSLContext sslContext;
+ private HttpClient client;
+ private boolean closed;
+
+ public OpenFgaHealthProbe(SSLContext sslContext) {
+ this.sslContext = sslContext;
+ }
+
+ private synchronized HttpClient client() {
+ if (closed) {
+ throw new IllegalStateException("The health probe is closed");
+ }
+ if (client == null) {
+ HttpClient.Builder builder =
HttpClient.newBuilder().connectTimeout(TIMEOUT);
+ if (sslContext != null) {
+ builder.sslContext(sslContext);
+ }
+ client = builder.build();
+ }
+ return client;
+ }
+
+ @Override
+ public synchronized void close() {
+ closed = true;
+ if (client != null) {
+ client.shutdownNow();
+ client = null;
+ }
+ }
/**
- * One client per distinct TLS configuration, because an {@link
SSLContext} can only be set when the client is
- * built. A probe that ignored it would fail its handshake against the
very server the authorization call reaches
- * happily - reporting DOWN, and with it an application that never becomes
ready.
- * <p/>
- * Keyed on identity and never evicted, which is bounded in practice: the
key is an {@code SSLContext} built from an
- * endpoint's {@code sslContextParameters}, and a deployment has one or
two of those, not one per exchange.
+ * The client this probe holds, without creating one. For the tests of the
probe's lifecycle.
*/
- private static final Map<SSLContext, HttpClient> TLS_CLIENTS = new
ConcurrentHashMap<>();
-
- private OpenFgaHealthProbe() {
+ synchronized HttpClient currentClient() {
+ return client;
}
- private static HttpClient clientFor(SSLContext sslContext) {
- if (sslContext == null) {
- return HTTP_CLIENT;
+ /**
+ * Sends the request and waits for the whole response, body included, for
at most {@link #TIMEOUT}. The timeout of
+ * the request itself only covers the wait for the response headers, so a
server that stalls while sending the body
+ * would otherwise hold the thread that runs the health check for as long
as it likes.
+ */
+ private <T> HttpResponse<T> sendBounded(HttpRequest request,
HttpResponse.BodyHandler<T> handler) throws Exception {
+ CompletableFuture<HttpResponse<T>> call = client().sendAsync(request,
handler);
+ try {
+ return call.get(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (TimeoutException e) {
+ call.cancel(true);
+ throw new HttpTimeoutException("Timed out waiting for the OpenFGA
health endpoint after " + TIMEOUT);
+ } catch (InterruptedException e) {
+ call.cancel(true);
+ throw e;
+ } catch (ExecutionException e) {
+ throw e.getCause() instanceof Exception cause ? cause : e;
}
- return TLS_CLIENTS.computeIfAbsent(sslContext,
- ctx ->
HttpClient.newBuilder().connectTimeout(TIMEOUT).sslContext(ctx).build());
}
/**
@@ -95,6 +127,12 @@ public final class OpenFgaHealthProbe {
*/
public static void probe(
HealthCheckResultBuilder builder, String apiUrl, String apiToken,
String storeId, SSLContext sslContext) {
+ try (OpenFgaHealthProbe probe = new OpenFgaHealthProbe(sslContext)) {
+ probe.probe(builder, apiUrl, apiToken, storeId);
+ }
+ }
+
+ public void probe(HealthCheckResultBuilder builder, String apiUrl, String
apiToken, String storeId) {
builder.detail("openfga.apiUrl", URISupport.sanitizeUri(apiUrl));
if (ObjectHelper.isNotEmpty(storeId)) {
builder.detail("openfga.storeId", storeId);
@@ -111,8 +149,7 @@ public final class OpenFgaHealthProbe {
}
try {
- HttpResponse<String> response
- = clientFor(sslContext).send(request.build(),
HttpResponse.BodyHandlers.ofString());
+ HttpResponse<String> response = sendBounded(request.build(),
HttpResponse.BodyHandlers.ofString());
if (response.statusCode() != 200) {
builder.down();
builder.message("OpenFGA answered its health endpoint with
HTTP " + response.statusCode());
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 a33c8d511799..8588e8f74608 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
@@ -111,6 +111,7 @@ public class OpenFgaProducer extends DefaultProducer {
protected void doStop() throws Exception {
if (healthCheckRepository != null && producerHealthCheck != null) {
healthCheckRepository.removeHealthCheck(producerHealthCheck);
+ producerHealthCheck.close();
producerHealthCheck = null;
}
super.doStop();
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducerHealthCheck.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducerHealthCheck.java
index d53b932e21f0..2ef1b2d36628 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducerHealthCheck.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/OpenFgaProducerHealthCheck.java
@@ -31,12 +31,12 @@ import org.apache.camel.util.URISupport;
* check probes the server's {@code /healthz} endpoint so that an unavailable
decision point is visible before traffic
* starts failing, rather than only in the error logs afterwards.
*/
-public class OpenFgaProducerHealthCheck extends AbstractHealthCheck {
+public class OpenFgaProducerHealthCheck extends AbstractHealthCheck implements
AutoCloseable {
private final String apiUrl;
private final String apiToken;
private final String storeId;
- private final SSLContext sslContext;
+ private final OpenFgaHealthProbe probe;
public OpenFgaProducerHealthCheck(String apiUrl, String apiToken, String
storeId, String id,
SSLContext sslContext) {
@@ -46,11 +46,16 @@ public class OpenFgaProducerHealthCheck extends
AbstractHealthCheck {
this.apiUrl = apiUrl;
this.apiToken = apiToken;
this.storeId = storeId;
- this.sslContext = sslContext;
+ this.probe = new OpenFgaHealthProbe(sslContext);
}
@Override
protected void doCall(HealthCheckResultBuilder builder, Map<String,
Object> options) {
- OpenFgaHealthProbe.probe(builder, apiUrl, apiToken, storeId,
sslContext);
+ probe.probe(builder, apiUrl, apiToken, storeId);
+ }
+
+ @Override
+ public void close() {
+ probe.close();
}
}
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicy.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicy.java
index be331da8d2d5..39eabc2352c8 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicy.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicy.java
@@ -246,6 +246,7 @@ public class OpenFgaSecurityPolicy implements
AuthorizationPolicy {
if (registry != null) {
registry.unregister(healthCheck);
}
+ healthCheck.close();
healthCheck = null;
}
diff --git
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicyHealthCheck.java
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicyHealthCheck.java
index 5db4c88d4b0c..21797321dc0a 100644
---
a/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicyHealthCheck.java
+++
b/components/camel-openfga/src/main/java/org/apache/camel/component/openfga/security/OpenFgaSecurityPolicyHealthCheck.java
@@ -33,12 +33,12 @@ import org.apache.camel.util.URISupport;
* unreachable server here fails every message outright, which is exactly the
condition worth surfacing before traffic
* arrives rather than after.
*/
-public class OpenFgaSecurityPolicyHealthCheck extends AbstractHealthCheck {
+public class OpenFgaSecurityPolicyHealthCheck extends AbstractHealthCheck
implements AutoCloseable {
private final String apiUrl;
private final String apiToken;
private final String storeId;
- private final SSLContext sslContext;
+ private final OpenFgaHealthProbe probe;
public OpenFgaSecurityPolicyHealthCheck(String apiUrl, String apiToken,
String storeId, String relation,
SSLContext sslContext) {
@@ -48,11 +48,16 @@ public class OpenFgaSecurityPolicyHealthCheck extends
AbstractHealthCheck {
this.apiUrl = apiUrl;
this.apiToken = apiToken;
this.storeId = storeId;
- this.sslContext = sslContext;
+ this.probe = new OpenFgaHealthProbe(sslContext);
}
@Override
protected void doCall(HealthCheckResultBuilder builder, Map<String,
Object> options) {
- OpenFgaHealthProbe.probe(builder, apiUrl, apiToken, storeId,
sslContext);
+ probe.probe(builder, apiUrl, apiToken, storeId);
+ }
+
+ @Override
+ public void close() {
+ probe.close();
}
}
diff --git
a/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaHealthProbeLifecycleTest.java
b/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaHealthProbeLifecycleTest.java
new file mode 100644
index 000000000000..3cc7f0ee14d4
--- /dev/null
+++
b/components/camel-openfga/src/test/java/org/apache/camel/component/openfga/OpenFgaHealthProbeLifecycleTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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 java.lang.reflect.Field;
+import java.net.http.HttpClient;
+import java.time.Duration;
+import java.util.List;
+
+import javax.net.ssl.SSLContext;
+
+import org.apache.camel.Producer;
+import
org.apache.camel.component.openfga.security.OpenFgaSecurityPolicyHealthCheck;
+import org.apache.camel.health.HealthCheck;
+import org.apache.camel.health.HealthCheckHelper;
+import org.apache.camel.health.WritableHealthCheckRepository;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.jsse.SSLContextParameters;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class OpenFgaHealthProbeLifecycleTest {
+
+ @Test
+ void closesProducerAndPolicyProbeClientsIndependently() throws Exception {
+ SSLContext ssl = SSLContext.getDefault();
+ try (OpenFgaProducerHealthCheck producer = new
OpenFgaProducerHealthCheck(
+ "http://localhost:1";, null, "scope", "endpoint", ssl);
+ OpenFgaSecurityPolicyHealthCheck policy = new
OpenFgaSecurityPolicyHealthCheck(
+ "http://localhost:1";, null, "scope", "viewer", ssl)) {
+ for (HealthCheck check : List.of(producer, policy)) {
+ check.setEnabled(true);
+
assertThat(check.call().getState()).isEqualTo(HealthCheck.State.DOWN);
+ }
+ HttpClient producerClient = client(producer);
+ HttpClient policyClient = client(policy);
+ producer.close();
+
+
await().atMost(Duration.ofSeconds(5)).until(producerClient::isTerminated);
+ assertThat(policyClient.isTerminated()).isFalse();
+ assertThat(producer.call().getError()).isPresent()
+ .get().isInstanceOf(IllegalStateException.class);
+ policy.close();
+
await().atMost(Duration.ofSeconds(5)).until(policyClient::isTerminated);
+ }
+ }
+
+ @Test
+ void releasesEachNewTlsConfigurationOnClose() throws Exception {
+ for (int i = 0; i < 5; i++) {
+ SSLContext ssl = SSLContext.getInstance("TLS");
+ ssl.init(null, null, null);
+ try (OpenFgaProducerHealthCheck check = new
OpenFgaProducerHealthCheck(
+ "http://localhost:1";, null, "scope", "endpoint", ssl)) {
+ check.setEnabled(true);
+ check.call();
+ HttpClient client = client(check);
+ check.close();
+ check.close();
+
await().atMost(Duration.ofSeconds(5)).until(client::isTerminated);
+
assertThat(check.call().getState()).isEqualTo(HealthCheck.State.DOWN);
+ assertThat(check.call().getError()).isPresent()
+ .get().isInstanceOf(IllegalStateException.class);
+ }
+ }
+ }
+
+ @Test
+ void producerRestartClosesTheOldClientAndCreatesAnIndependentProbe()
throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ context.getComponent("openfga",
OpenFgaComponent.class).setHealthCheckProducerEnabled(true);
+ context.getRegistry().bind("probeTls", new SSLContextParameters());
+ context.start();
+ Producer producer = context.getEndpoint(
+
"openfga:check?apiUrl=http://localhost:1&storeId=01HQMVAJXYZ0000000000000&relation=viewer&user=user:anne&object=document:budget&sslContextParameters=#probeTls";)
+ .createProducer();
+ context.addService(producer, true);
+ try {
+ WritableHealthCheckRepository repository =
HealthCheckHelper.getHealthCheckRepository(
+ context, "producers",
WritableHealthCheckRepository.class);
+ assertThat(repository).isNotNull();
+ repository.setEnabled(true);
+ HealthCheck first = repository.stream()
+ .filter(check ->
check.getId().startsWith("producer:openfga-")).findFirst().orElseThrow();
+ first.call();
+ HttpClient firstClient = client(first);
+ producer.stop();
+
await().atMost(Duration.ofSeconds(5)).until(firstClient::isTerminated);
+ assertThat(repository.stream().filter(check ->
check.getId().startsWith("producer:openfga-"))).isEmpty();
+
+ producer.start();
+ HealthCheck restarted = repository.stream()
+ .filter(check ->
check.getId().startsWith("producer:openfga-")).findFirst().orElseThrow();
+ assertThat(restarted).isNotSameAs(first);
+ restarted.call();
+ HttpClient restartedClient = client(restarted);
+ assertThat(restartedClient).isNotSameAs(firstClient);
+ context.stop();
+
await().atMost(Duration.ofSeconds(5)).until(restartedClient::isTerminated);
+ } finally {
+ producer.stop();
+ }
+ }
+ }
+
+ private static HttpClient client(HealthCheck check) throws Exception {
+ Field probe = check.getClass().getDeclaredField("probe");
+ probe.setAccessible(true);
+ // the probe exposes its client to this package, so a rename there is
a compile error, not a surprise
+ return ((OpenFgaHealthProbe) probe.get(check)).currentClient();
+ }
+}
