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 dd1b3a96f807 CAMEL-26743: camel-sql -
JdbcOrphanLockAwareIdempotentRepository: the keep-alive deadlocks with add and
remove on a small connection pool (#29437)
dd1b3a96f807 is described below
commit dd1b3a96f8073cd641f27aece61257020c506ea3
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 9 15:39:25 2026 +0530
CAMEL-26743: camel-sql - JdbcOrphanLockAwareIdempotentRepository: the
keep-alive deadlocks with add and remove on a small connection pool (#29437)
* CAMEL-26743: camel-sql - JdbcOrphanLockAwareIdempotentRepository: the
keep-alive deadlocks with add and remove on a small connection pool
* CAMEL-26743: camel-sql - JdbcOrphanLockAwareIdempotentRepository: keep a
key alive only after its row is committed
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../JdbcOrphanLockAwareIdempotentRepository.java | 143 ++++++----
...LockAwareIdempotentRepositoryKeepAliveTest.java | 304 +++++++++++++++++++++
...ckAwareIdempotentRepositoryTransactionTest.java | 154 +++++++++++
3 files changed, 549 insertions(+), 52 deletions(-)
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
index 10bcaee14429..c7a9d1b741d3 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
@@ -17,13 +17,12 @@
package org.apache.camel.processor.idempotent.jdbc;
import java.sql.Timestamp;
-import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.StampedLock;
import java.util.stream.Collectors;
import javax.sql.DataSource;
@@ -31,13 +30,15 @@ import javax.sql.DataSource;
import org.apache.camel.CamelContext;
import org.apache.camel.spi.ExecutorServiceManager;
import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
/**
* Implementation of {@link AbstractJdbcMessageIdRepository} which handles
orphan locks resulting from jvm crash.
*
* When an instance of the application acquires a lock on the idempotent
repository, the lock attributes are added to a
- * HashSet. While the lock is help by the instance, the instance keeps
updating the createdAt column with the current
+ * set. While the lock is held by the instance, the instance keeps updating
the createdAt column with the current
* timestamp indicating the instance holding the lock is active.
*
* A lock is granted to an instance if either the entry for the lock
attributes do not exists in the
@@ -49,9 +50,9 @@ public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdReposi
protected static final String DEFAULT_UPDATE_TIMESTAMP_QUERY
= "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =? WHERE
processorName =? AND messageId = ?";
- private final StampedLock sl = new StampedLock();
-
- private final Set<ProcessorNameAndMessageId> processorNameMessageIdSet =
new HashSet<>();
+ // a concurrent set, so no lock is held while SQL runs: add and remove
hold a connection of the pool while they
+ // run, and the keep-alive needs one too, so a lock held across SQL could
deadlock with a small pool
+ private final Set<ProcessorNameAndMessageId> processorNameMessageIdSet =
ConcurrentHashMap.newKeySet();
private ExecutorServiceManager executorServiceManager;
@@ -104,42 +105,77 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
@Override
protected int delete(String key) {
- long stamp = sl.writeLock();
- try {
- int result = super.delete(key);
- processorNameMessageIdSet.remove(new
ProcessorNameAndMessageId(processorName, key));
- return result;
- } finally {
- sl.unlockWrite(stamp);
+ int result = super.delete(key);
+ ProcessorNameAndMessageId id = new
ProcessorNameAndMessageId(processorName, key);
+ processorNameMessageIdSet.remove(id);
+ // an insert of the same key earlier in this transaction keeps it
alive only after the commit: cancel that
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ for (TransactionSynchronization synchronization :
TransactionSynchronizationManager.getSynchronizations()) {
+ if (synchronization instanceof KeepAliveAfterCommit pending &&
pending.repository == this
+ && pending.id.equals(id)) {
+ pending.cancelled = true;
+ }
+ }
}
-
+ return result;
}
@Override
protected int insert(String key) {
Timestamp currentTimestamp = new Timestamp(System.currentTimeMillis());
- long stamp = sl.writeLock();
- try {
- if (jdbcTemplate.queryForObject(getQueryString(), Integer.class,
processorName, key) == 0) {
- int result = jdbcTemplate.update(getInsertString(),
processorName, key, currentTimestamp);
- processorNameMessageIdSet.add(new
ProcessorNameAndMessageId(processorName, key));
- return result;
- } else {
- // Row exists — try to claim orphan lock with conditional
UPDATE.
- // Only succeeds if createdAt is older than lockMaxAge (lock
is truly orphaned).
- // Returns 0 if the lock is held by an active instance,
preventing the race in CAMEL-23294.
- String orphanLockRecoverQueryString =
getUpdateTimestampQuery() + " AND createdAt < ?";
- Timestamp xMillisAgo = new
Timestamp(System.currentTimeMillis() - lockMaxAgeMillis);
- int result
- = jdbcTemplate.update(orphanLockRecoverQueryString,
currentTimestamp, processorName, key, xMillisAgo);
- if (result > 0) {
- log.debug("Orphan lock seized for key: {}", key);
- processorNameMessageIdSet.add(new
ProcessorNameAndMessageId(processorName, key));
- }
- return result;
+ if (jdbcTemplate.queryForObject(getQueryString(), Integer.class,
processorName, key) == 0) {
+ int result = jdbcTemplate.update(getInsertString(), processorName,
key, currentTimestamp);
+ keepAliveAfterCommit(key);
+ return result;
+ } else {
+ // Row exists — try to claim orphan lock with conditional UPDATE.
+ // Only succeeds if createdAt is older than lockMaxAge (lock is
truly orphaned).
+ // Returns 0 if the lock is held by an active instance, preventing
the race in CAMEL-23294.
+ String orphanLockRecoverQueryString = getUpdateTimestampQuery() +
" AND createdAt < ?";
+ Timestamp xMillisAgo = new Timestamp(System.currentTimeMillis() -
lockMaxAgeMillis);
+ int result = jdbcTemplate.update(orphanLockRecoverQueryString,
currentTimestamp, processorName, key, xMillisAgo);
+ if (result > 0) {
+ log.debug("Orphan lock seized for key: {}", key);
+ keepAliveAfterCommit(key);
+ }
+ return result;
+ }
+ }
+
+ /**
+ * Keeps the lease of the key alive once the transaction that inserted or
seized its row commits. When that
+ * transaction rolls back (for example in a transacted route with
removeOnFailure=false), the row is not there, so
+ * the key must not be kept alive: another node may insert it later, and
this node would keep its lease alive.
+ */
+ private void keepAliveAfterCommit(String key) {
+ ProcessorNameAndMessageId id = new
ProcessorNameAndMessageId(processorName, key);
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
KeepAliveAfterCommit(this, id));
+ } else {
+ processorNameMessageIdSet.add(id);
+ }
+ }
+
+ /**
+ * Adds a key to the keys kept alive after the commit of the transaction
that inserted or seized its row, unless the
+ * key was deleted later in that transaction.
+ */
+ private static final class KeepAliveAfterCommit implements
TransactionSynchronization {
+ private final JdbcOrphanLockAwareIdempotentRepository repository;
+ private final ProcessorNameAndMessageId id;
+ // set and read by the thread of the transaction
+ private boolean cancelled;
+
+ private KeepAliveAfterCommit(JdbcOrphanLockAwareIdempotentRepository
repository, ProcessorNameAndMessageId id) {
+ this.repository = repository;
+ this.id = id;
+ }
+
+ @Override
+ public void afterCommit() {
+ if (!cancelled) {
+ repository.processorNameMessageIdSet.add(id);
}
- } finally {
- sl.unlockWrite(stamp);
}
}
@@ -181,30 +217,33 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
@Override
protected int delete() {
- long stamp = sl.writeLock();
- try {
- int result = super.delete();
- processorNameMessageIdSet.clear();
- return result;
- } finally {
- sl.unlockWrite(stamp);
+ int result = super.delete();
+ processorNameMessageIdSet.clear();
+ // and the inserts earlier in this transaction are not kept alive
after the commit either
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ for (TransactionSynchronization synchronization :
TransactionSynchronizationManager.getSynchronizations()) {
+ if (synchronization instanceof KeepAliveAfterCommit pending &&
pending.repository == this) {
+ pending.cancelled = true;
+ }
+ }
}
-
+ return result;
}
void keepAlive() {
- Timestamp currentTimestamp = new Timestamp(System.currentTimeMillis());
- long stamp = sl.readLock();
try {
- List<Object[]> args = processorNameMessageIdSet.stream()
- .map(processorNameMessageId -> new Object[] {
- currentTimestamp,
processorNameMessageId.processorName, processorNameMessageId.messageId })
- .collect(Collectors.toList());
- transactionTemplate.execute(status ->
jdbcTemplate.batchUpdate(getUpdateTimestampQuery(), args));
+ transactionTemplate.execute(status -> {
+ // take the timestamp and the keys inside the transaction,
after the connection was obtained (with the
+ // DataSourceTransactionManager), as waiting for it can take a
while
+ Timestamp currentTimestamp = new
Timestamp(System.currentTimeMillis());
+ List<Object[]> args = processorNameMessageIdSet.stream()
+ .map(processorNameMessageId -> new Object[] {
+ currentTimestamp,
processorNameMessageId.processorName, processorNameMessageId.messageId })
+ .collect(Collectors.toList());
+ return jdbcTemplate.batchUpdate(getUpdateTimestampQuery(),
args);
+ });
} catch (Exception e) {
log.error("failed updating createdAt in keepAlive due to ", e);
- } finally {
- sl.unlockRead(stamp);
}
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryKeepAliveTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryKeepAliveTest.java
new file mode 100644
index 000000000000..81bab6affcee
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryKeepAliveTest.java
@@ -0,0 +1,304 @@
+/*
+ * 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.processor.idempotent.jdbc;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DelegatingDataSource;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The keep-alive must not wait for the add or remove of a key, nor make them
wait, while either holds a connection of
+ * the pool. With a lock held across the SQL, a keep-alive that waits for the
only connection while add or remove,
+ * holding that connection, waits for the lock deadlocks the node; and a
keep-alive that waits while add runs a slow
+ * statement (such as one waiting for a row lock of another node) lets the
leases of the node expire.
+ */
+class JdbcOrphanLockAwareIdempotentRepositoryKeepAliveTest {
+
+ private static final String PROCESSOR_NAME = "APP_1";
+ private static final String HELD = "HELD";
+ private static final String KEY = "KEY";
+ private static final long LOCK_MAX_AGE_MILLIS = 300_000L;
+ private static final long WAIT_SECONDS = 10;
+
+ private EmbeddedDatabase database;
+ private JdbcTemplate jdbcTemplate;
+ private JdbcOrphanLockAwareIdempotentRepository repository;
+ private Pool pool;
+ private Timestamp staleCreatedAt;
+
+ @BeforeEach
+ void setup() {
+ database = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcTemplate = new JdbcTemplate(database);
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (pool != null) {
+ // a thread held by the pool is let go, also when the test failed
before it did so
+ pool.release.countDown();
+ }
+ if (repository != null) {
+ repository.shutdown();
+ }
+ database.shutdown();
+ }
+
+ private void start(int connections) {
+ pool = new Pool(database, connections);
+ repository = new JdbcOrphanLockAwareIdempotentRepository(pool,
PROCESSOR_NAME, new DefaultCamelContext());
+ repository.setLockMaxAgeMillis(LOCK_MAX_AGE_MILLIS);
+ // the test calls keepAlive() itself
+ repository.setLockKeepAliveIntervalMillis(LOCK_MAX_AGE_MILLIS / 2);
+ repository.start();
+ assertTrue(repository.add(HELD));
+ // the lease of the held key is old, so a keep-alive that runs is seen
+ staleCreatedAt = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS / 2);
+ jdbcTemplate.update("UPDATE CAMEL_MESSAGEPROCESSED SET createdAt = ?
WHERE messageId = ?", staleCreatedAt, HELD);
+ }
+
+ @Test
+ void testKeepAliveWaitingForTheOnlyConnectionDoesNotBlockAdd() throws
Exception {
+ start(1);
+ AtomicReference<Object> added = new AtomicReference<>();
+ Thread adder = worker("adder", added, () -> repository.add(KEY));
+
+ keepAliveWhileHoldingTheOnlyConnection(adder);
+
+ assertEquals(Boolean.TRUE, added.get());
+ assertEquals(2, repository.getProcessorNameMessageIdSet().size());
+ }
+
+ @Test
+ void testKeepAliveWaitingForTheOnlyConnectionDoesNotBlockRemove() throws
Exception {
+ start(1);
+ assertTrue(repository.add(KEY));
+ AtomicReference<Object> removed = new AtomicReference<>();
+ Thread remover = worker("remover", removed, () ->
repository.remove(KEY));
+
+ keepAliveWhileHoldingTheOnlyConnection(remover);
+
+ assertEquals(Boolean.TRUE, removed.get());
+ assertEquals(1, repository.getProcessorNameMessageIdSet().size());
+ }
+
+ /**
+ * The worker borrows the only connection and is held there; the
keep-alive then waits for the connection, and the
+ * worker goes on. Both must finish, and the keep-alive must refresh the
held key.
+ */
+ private void keepAliveWhileHoldingTheOnlyConnection(Thread worker) throws
Exception {
+ CountDownLatch keepAliveWaits = new CountDownLatch(1);
+ pool.holdAfterBorrow(worker);
+ pool.onWait = keepAliveWaits::countDown;
+ worker.start();
+ assertTrue(pool.reached.await(WAIT_SECONDS, TimeUnit.SECONDS));
+
+ Thread keeper = new Thread(repository::keepAlive, "keep-alive");
+ keeper.start();
+ assertTrue(keepAliveWaits.await(WAIT_SECONDS, TimeUnit.SECONDS));
+ pool.release.countDown();
+
+ try {
+ worker.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ keeper.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ assertFalse(worker.isAlive(), () -> worker.getName() + " is
blocked: " + stackOf(worker));
+ assertFalse(keeper.isAlive(), () -> "the keep-alive is blocked: "
+ stackOf(keeper));
+ } finally {
+ // on a deadlock, end the wait of the keep-alive for the
connection, so both threads end
+ keeper.interrupt();
+ keeper.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ worker.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ }
+ assertTrue(createdAt(HELD).after(staleCreatedAt), "the keep-alive did
not refresh the held key");
+ }
+
+ @Test
+ void testKeepAliveDoesNotWaitForASlowInsert() throws Exception {
+ start(2);
+ AtomicReference<Object> added = new AtomicReference<>();
+ Thread adder = worker("adder", added, () -> repository.add(KEY));
+ // the INSERT of the add is held, as a statement that waits for a row
lock of another node would be
+ pool.holdAtStatement(adder, "INSERT");
+ adder.start();
+ assertTrue(pool.reached.await(WAIT_SECONDS, TimeUnit.SECONDS));
+
+ Thread keeper = new Thread(repository::keepAlive, "keep-alive");
+ try {
+ keeper.start();
+ keeper.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ assertFalse(keeper.isAlive(), () -> "the keep-alive waits for the
INSERT: " + stackOf(keeper));
+ assertTrue(createdAt(HELD).after(staleCreatedAt), "the keep-alive
did not refresh the held key");
+ } finally {
+ pool.release.countDown();
+ adder.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ keeper.join(TimeUnit.SECONDS.toMillis(WAIT_SECONDS));
+ }
+ assertFalse(adder.isAlive());
+ assertEquals(Boolean.TRUE, added.get());
+ }
+
+ /**
+ * A thread that calls the repository and keeps what it returned, or what
it threw.
+ */
+ private static Thread worker(String name, AtomicReference<Object> result,
Supplier<Boolean> call) {
+ return new Thread(() -> {
+ try {
+ result.set(call.get());
+ } catch (Throwable e) {
+ result.set(e);
+ }
+ }, name);
+ }
+
+ private Timestamp createdAt(String key) {
+ return jdbcTemplate.queryForObject("SELECT createdAt FROM
CAMEL_MESSAGEPROCESSED WHERE messageId = ?",
+ Timestamp.class, key);
+ }
+
+ private static String stackOf(Thread thread) {
+ StringBuilder sb = new StringBuilder(thread.getState().toString());
+ for (StackTraceElement element : thread.getStackTrace()) {
+ sb.append("\n\tat ").append(element);
+ }
+ return sb.toString();
+ }
+
+ /**
+ * A pool of a fixed number of connections to the embedded database, which
waits for a connection as long as it
+ * takes, with a point where it holds one thread: after it borrowed a
connection, or before it runs a statement.
+ */
+ private static final class Pool extends DelegatingDataSource {
+
+ private final Semaphore free;
+ private final CountDownLatch reached = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final AtomicReference<Thread> held = new AtomicReference<>();
+ private volatile String heldAtStatement;
+ private volatile Runnable onWait = () -> {
+ };
+
+ Pool(EmbeddedDatabase database, int connections) {
+ super(database);
+ this.free = new Semaphore(connections);
+ }
+
+ void holdAfterBorrow(Thread thread) {
+ held.set(thread);
+ }
+
+ void holdAtStatement(Thread thread, String sqlPrefix) {
+ heldAtStatement = sqlPrefix;
+ held.set(thread);
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ if (!free.tryAcquire()) {
+ onWait.run();
+ try {
+ free.acquire();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new SQLException("Interrupted while waiting for a
connection", e);
+ }
+ }
+ Connection connection;
+ try {
+ connection = super.getConnection();
+ } catch (SQLException | RuntimeException e) {
+ free.release();
+ throw e;
+ }
+ if (heldAtStatement == null) {
+ hold();
+ }
+ return pooled(connection);
+ }
+
+ private void hold() {
+ if (held.compareAndSet(Thread.currentThread(), null)) {
+ reached.countDown();
+ try {
+ // bounded, so a failed test cannot leave the thread
waiting forever
+ release.await(WAIT_SECONDS * 3, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ /**
+ * The connection, which goes back to the pool when it is closed.
+ */
+ private Connection pooled(Connection connection) {
+ AtomicBoolean closed = new AtomicBoolean();
+ return (Connection)
Proxy.newProxyInstance(getClass().getClassLoader(), new Class<?>[] {
Connection.class },
+ (proxy, method, args) -> {
+ if ("close".equals(method.getName())) {
+ if (closed.compareAndSet(false, true)) {
+ try {
+ connection.close();
+ } finally {
+ free.release();
+ }
+ }
+ return null;
+ }
+ if ("equals".equals(method.getName())) {
+ return proxy == args[0];
+ }
+ if ("hashCode".equals(method.getName())) {
+ return System.identityHashCode(proxy);
+ }
+ if ("prepareStatement".equals(method.getName()) &&
heldAtStatement != null
+ && ((String)
args[0]).startsWith(heldAtStatement)) {
+ hold();
+ }
+ try {
+ return method.invoke(connection, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ });
+ }
+ }
+}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryTransactionTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryTransactionTest.java
new file mode 100644
index 000000000000..03529d1dd2e0
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryTransactionTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.processor.idempotent.jdbc;
+
+import java.sql.Timestamp;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import
org.apache.camel.processor.idempotent.jdbc.JdbcOrphanLockAwareIdempotentRepository.ProcessorNameAndMessageId;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A key is kept alive only once the transaction that inserted or seized its
row commits. In a transaction of the caller
+ * that rolls back (for example a transacted route with
removeOnFailure=false), the row is not there, and the node must
+ * not keep the key alive: another node may insert it later, and this node
would keep that node's lease alive.
+ */
+class JdbcOrphanLockAwareIdempotentRepositoryTransactionTest {
+
+ private static final String PROCESSOR_NAME = "APP_1";
+ private static final String KEY = "KEY";
+ private static final long LOCK_MAX_AGE_MILLIS = 300_000L;
+ private static final ProcessorNameAndMessageId ID = new
ProcessorNameAndMessageId(PROCESSOR_NAME, KEY);
+
+ private EmbeddedDatabase database;
+ private JdbcTemplate jdbcTemplate;
+ private JdbcOrphanLockAwareIdempotentRepository repository;
+
+ @BeforeEach
+ void setup() {
+ database = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcTemplate = new JdbcTemplate(database);
+ repository = new JdbcOrphanLockAwareIdempotentRepository(database,
PROCESSOR_NAME, new DefaultCamelContext());
+ repository.setLockMaxAgeMillis(LOCK_MAX_AGE_MILLIS);
+ // the keep-alive task does not run during the test
+ repository.setLockKeepAliveIntervalMillis(LOCK_MAX_AGE_MILLIS / 2);
+ repository.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ repository.shutdown();
+ database.shutdown();
+ }
+
+ @Test
+ void testAddInACallerTransactionThatRollsBackIsNotKeptAlive() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ status.setRollbackOnly();
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testAddInACallerTransactionIsKeptAliveAfterTheCommit() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+
assertFalse(repository.getProcessorNameMessageIdSet().contains(ID));
+ });
+
+ assertEquals(1, rows());
+ assertTrue(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testAddIsKeptAlive() {
+ assertTrue(repository.add(KEY));
+
+ assertTrue(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testAddAndRemoveInOneCallerTransactionIsNotKeptAlive() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ assertTrue(repository.remove(KEY));
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testAddAndClearInOneCallerTransactionIsNotKeptAlive() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ repository.clear();
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testOrphanLockSeizedInACallerTransactionThatRollsBackIsNotKeptAlive()
{
+ Timestamp orphaned = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS * 2);
+ jdbcTemplate.update("INSERT INTO CAMEL_MESSAGEPROCESSED
(processorName, messageId, createdAt) VALUES (?, ?, ?)",
+ PROCESSOR_NAME, KEY, orphaned);
+
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ status.setRollbackOnly();
+ });
+
+ assertEquals(orphaned, jdbcTemplate.queryForObject(
+ "SELECT createdAt FROM CAMEL_MESSAGEPROCESSED WHERE
processorName = ? AND messageId = ?",
+ Timestamp.class, PROCESSOR_NAME, KEY));
+ assertFalse(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ @Test
+ void testOrphanLockSeizedInACallerTransactionIsKeptAliveAfterTheCommit() {
+ Timestamp orphaned = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS * 2);
+ jdbcTemplate.update("INSERT INTO CAMEL_MESSAGEPROCESSED
(processorName, messageId, createdAt) VALUES (?, ?, ?)",
+ PROCESSOR_NAME, KEY, orphaned);
+
+ repository.transactionTemplate.executeWithoutResult(status ->
assertTrue(repository.add(KEY)));
+
+ assertTrue(repository.getProcessorNameMessageIdSet().contains(ID));
+ }
+
+ private int rows() {
+ return jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
CAMEL_MESSAGEPROCESSED WHERE processorName = ?",
+ Integer.class, PROCESSOR_NAME);
+ }
+}
