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 5c2e5ffb226c CAMEL-26830: Serialize WASM pool shutdown and policy
return (#29633)
5c2e5ffb226c is described below
commit 5c2e5ffb226cc164cfc434b8047d4bdc227de9e6
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Nov 11 19:50:44 2026 +0100
CAMEL-26830: Serialize WASM pool shutdown and policy return (#29633)
camel-opa: reject queued WASM borrows after the policy pool is closed, and
serialize pool shutdown with
returning a policy to the pool.
Co-authored-by: Codex <[email protected]>
Signed-off-by: Andrea Cosentino <[email protected]>
---
.../camel/component/opa/OpaWasmPolicyPool.java | 34 +++++--
.../camel/component/opa/OpaWasmPolicyPoolTest.java | 103 +++++++++++++++++++++
2 files changed, 127 insertions(+), 10 deletions(-)
diff --git
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaWasmPolicyPool.java
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaWasmPolicyPool.java
index c31b34ffee83..f7617f5d59bb 100644
---
a/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaWasmPolicyPool.java
+++
b/components/camel-opa/src/main/java/org/apache/camel/component/opa/OpaWasmPolicyPool.java
@@ -16,8 +16,8 @@
*/
package org.apache.camel.component.opa;
+import java.util.ArrayDeque;
import java.util.Deque;
-import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -43,10 +43,10 @@ import com.styra.opa.wasm.OpaPolicy;
*/
final class OpaWasmPolicyPool implements AutoCloseable {
- private final Deque<OpaPolicy> idle = new ConcurrentLinkedDeque<>();
+ private final Deque<OpaPolicy> idle = new ArrayDeque<>();
private final Semaphore permits;
private final Supplier<OpaPolicy> factory;
- private final AtomicBoolean closed = new AtomicBoolean();
+ private volatile boolean closed;
private final int size;
private final long borrowTimeout;
@@ -64,7 +64,7 @@ final class OpaWasmPolicyPool implements AutoCloseable {
* @throws InterruptedException if the thread was interrupted while
waiting; the interrupt flag is left set
*/
Lease borrow() throws TimeoutException, InterruptedException {
- if (closed.get()) {
+ if (closed) {
throw new IllegalStateException("The policy pool is closed");
}
boolean acquired;
@@ -83,22 +83,34 @@ final class OpaWasmPolicyPool implements AutoCloseable {
+ " exchanges concurrently than that,
or borrowTimeout if the policy itself"
+ " legitimately takes this long to
evaluate");
}
- OpaPolicy policy;
try {
- policy = idle.pollFirst();
+ OpaPolicy policy;
+ synchronized (idle) {
+ if (closed) {
+ throw new IllegalStateException("The policy pool is
closed");
+ }
+ policy = idle.pollFirst();
+ }
if (policy == null) {
+ // Construction can be expensive, so do not prevent another
lease returning or the pool closing.
policy = factory.get();
}
+ synchronized (idle) {
+ if (closed) {
+ throw new IllegalStateException("The policy pool is
closed");
+ }
+ return new Lease(policy);
+ }
} catch (RuntimeException e) {
permits.release();
throw e;
}
- return new Lease(policy);
}
@Override
public void close() {
- if (closed.compareAndSet(false, true)) {
+ synchronized (idle) {
+ closed = true;
idle.clear();
}
// threads already waiting are not woken here: they are bounded by
borrowTimeout, and a shutdown that
@@ -136,8 +148,10 @@ final class OpaWasmPolicyPool implements AutoCloseable {
return;
}
try {
- if (!closed.get()) {
- idle.offerFirst(policy);
+ synchronized (idle) {
+ if (!closed) {
+ idle.offerFirst(policy);
+ }
}
} finally {
permits.release();
diff --git
a/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaWasmPolicyPoolTest.java
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaWasmPolicyPoolTest.java
index 2a053ad8895e..9f9577ad9c09 100644
---
a/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaWasmPolicyPoolTest.java
+++
b/components/camel-opa/src/test/java/org/apache/camel/component/opa/OpaWasmPolicyPoolTest.java
@@ -28,6 +28,7 @@ import org.junit.jupiter.api.Timeout;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
import static org.mockito.Mockito.mock;
/**
@@ -161,6 +162,108 @@ class OpaWasmPolicyPoolTest {
.hasMessageContaining("closed");
}
+ @Test
+ @Timeout(30)
+ void refusesAQueuedBorrowAfterThePoolCloses() throws Exception {
+ AtomicInteger built = new AtomicInteger();
+ OpaWasmPolicyPool pool = new OpaWasmPolicyPool(() -> {
+ built.incrementAndGet();
+ return mock(OpaPolicy.class);
+ }, 1, 10_000);
+ OpaWasmPolicyPool.Lease held = pool.borrow();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread waiter = new Thread(() -> {
+ try (OpaWasmPolicyPool.Lease ignored = pool.borrow()) {
+ failure.set(new AssertionError("The closed pool handed out a
lease"));
+ } catch (Exception e) {
+ failure.set(e);
+ }
+ });
+ try {
+ waiter.start();
+ await().atMost(10, TimeUnit.SECONDS).until(() -> waiter.getState()
== Thread.State.TIMED_WAITING);
+
+ pool.close();
+ held.close();
+ waiter.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertThat(waiter.isAlive()).isFalse();
+
assertThat(failure.get()).isInstanceOf(IllegalStateException.class).hasMessageContaining("closed");
+ assertThat(built).as("the queued borrower did not create another
policy after close").hasValue(1);
+ } finally {
+ pool.close();
+ held.close();
+ waiter.interrupt();
+ waiter.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ }
+
+ @Test
+ @Timeout(30)
+ void refusesALeaseWhenThePoolClosesDuringConstruction() throws Exception {
+ CountDownLatch constructing = new CountDownLatch(1);
+ CountDownLatch finishConstruction = new CountDownLatch(1);
+ OpaWasmPolicyPool pool = new OpaWasmPolicyPool(() -> {
+ constructing.countDown();
+ try {
+ if (!finishConstruction.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timed out waiting to
finish policy construction");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ return mock(OpaPolicy.class);
+ }, 1, 10_000);
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread borrower = new Thread(() -> {
+ try (OpaWasmPolicyPool.Lease ignored = pool.borrow()) {
+ failure.set(new AssertionError("The closed pool handed out a
lease"));
+ } catch (Exception e) {
+ failure.set(e);
+ }
+ });
+ try {
+ borrower.start();
+ assertThat(constructing.await(10, TimeUnit.SECONDS)).isTrue();
+
+ pool.close();
+ finishConstruction.countDown();
+ borrower.join(TimeUnit.SECONDS.toMillis(10));
+
+ assertThat(borrower.isAlive()).isFalse();
+
assertThat(failure.get()).isInstanceOf(IllegalStateException.class).hasMessageContaining("closed");
+ } finally {
+ pool.close();
+ finishConstruction.countDown();
+ borrower.interrupt();
+ borrower.join(TimeUnit.SECONDS.toMillis(10));
+ }
+ }
+
+ @Test
+ @Timeout(30)
+ void releasesThePermitWhenConstructionFails() throws Exception {
+ AtomicInteger built = new AtomicInteger();
+ OpaWasmPolicyPool pool = new OpaWasmPolicyPool(() -> {
+ if (built.incrementAndGet() == 1) {
+ throw new IllegalArgumentException("Invalid policy");
+ }
+ return mock(OpaPolicy.class);
+ }, 1, 150);
+ try {
+
assertThatThrownBy(pool::borrow).isInstanceOf(IllegalArgumentException.class);
+ try (OpaWasmPolicyPool.Lease held = pool.borrow()) {
+
assertThatThrownBy(pool::borrow).isInstanceOf(TimeoutException.class);
+ }
+ try (OpaWasmPolicyPool.Lease held = pool.borrow()) {
+ assertThat(built).as("the permit was released exactly once and
the instance was reused").hasValue(2);
+ }
+ } finally {
+ pool.close();
+ }
+ }
+
@Test
@Timeout(30)
void leavesTheInterruptFlagSetWhenInterrupted() throws Exception {
