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 4f2a45a68ab9 CAMEL-26868: camel-sql - JdbcCachedMessageIdRepository: a
key removed in a transaction of the caller is removed from the cache again when
it commits (#29669)
4f2a45a68ab9 is described below
commit 4f2a45a68ab92367973196f60d6bc7fb017330b4
Author: allthingssecurity <[email protected]>
AuthorDate: Thu Nov 12 17:33:03 2026 +0530
CAMEL-26868: camel-sql - JdbcCachedMessageIdRepository: a key removed in a
transaction of the caller is removed from the cache again when it commits
(#29669)
* CAMEL-26868: camel-sql - JdbcCachedMessageIdRepository: a key removed in
a transaction of the caller is removed from the cache again when it commits
JdbcCachedMessageIdRepository.remove (and clear) removes the key from the
cache before and after the delete. In a transaction of the caller on the
same DataSource (or JTA), for example a direct remove or clear in a
transaction, or the idempotent consumer removing the key of a failure that
a dead letter channel handled inside an enclosing transaction that
commits, the delete is only committed later. Until then another thread
still sees the row: its add of the key is a duplicate and caches the key.
When the transaction commits, the row is gone but the key stays in the
cache, so every later add of that key is rejected as a duplicate without a
row, until the cache is reloaded.
remove (when it deleted a row) and clear now also remove the key (or
every key) from the cache when the transaction of the caller commits (a
transaction synchronization, as the add uses since CAMEL-26770). Without
such a transaction the delete has committed when remove returns, as
before. A remove and an add of the key in one transaction still leave the
key cached: the synchronizations run in the order they were registered.
This narrows the window, it does not close it: an add whose SELECT saw the
row before the commit and that caches the key after the eviction (without
a transaction of its own, or with one that commits later) still caches it.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-26868: camel-sql - JdbcCachedMessageIdRepository: an add that read
the row before a remove or clear deleted it does not keep the key cached
Review of #29669: close the remaining window instead of a follow-up. An add
on
another thread whose SELECT saw the row before the delete committed, and
that
cached the key after the eviction ran, left the key cached without a row.
- A counter of evictions: every eviction after a delete (in remove when it
deleted a row, in clear, and when a transaction of the caller that deleted
commits) bumps it before the key leaves the cache. An add reads it before
its
SELECT, merges the key, and removes it again if the counter moved: either
the
eviction runs after the merge, or the add sees the change. An add that
runs
alongside a remove or clear of any key is not cached (one more SELECT
later).
- The cache hit of add is one computeIfPresent, so a key evicted between the
read and the increment is not put back.
- An add of a key later in the same transaction as its remove cancels the
eviction of that remove at the commit (the row is there again).
- reload() starts with an empty cache if a remove or clear ran while it read
the keys.
- sql-component.adoc: one sentence on remove and clear in a transaction.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/sql-component.adoc | 4 +
.../camel-sql/src/main/docs/sql-component.adoc | 4 +
.../jdbc/JdbcCachedMessageIdRepository.java | 124 +++++++++++++--
...bcCachedMessageIdRepositoryTransactionTest.java | 166 +++++++++++++++++++++
4 files changed, 283 insertions(+), 15 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
index 9ba79a81b105..8bfd8c543df7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
@@ -784,6 +784,10 @@ that exchange is rolled back once before its redelivery is
filtered as a
duplicate. The `JdbcMessageIdRepository`, and several nodes sharing the
table, behave the same way.
+A `remove` or `clear` in a transaction takes the key (or the keys) out of
+the cache again when that transaction commits, and an add that read the
+row before it was deleted does not keep the key in the cache.
+
=== Using the JDBC based aggregation repository
`JdbcAggregationRepository` is an `AggregationRepository` which on the
diff --git a/components/camel-sql/src/main/docs/sql-component.adoc
b/components/camel-sql/src/main/docs/sql-component.adoc
index 9ba79a81b105..8bfd8c543df7 100644
--- a/components/camel-sql/src/main/docs/sql-component.adoc
+++ b/components/camel-sql/src/main/docs/sql-component.adoc
@@ -784,6 +784,10 @@ that exchange is rolled back once before its redelivery is
filtered as a
duplicate. The `JdbcMessageIdRepository`, and several nodes sharing the
table, behave the same way.
+A `remove` or `clear` in a transaction takes the key (or the keys) out of
+the cache again when that transaction commits, and an add that read the
+row before it was deleted does not keep the key in the cache.
+
=== Using the JDBC based aggregation repository
`JdbcAggregationRepository` is an `AggregationRepository` which on the
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
index 546d446bbfc7..0ae7cc6fb82b 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
@@ -19,6 +19,7 @@ package org.apache.camel.processor.idempotent.jdbc;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import javax.sql.DataSource;
@@ -39,6 +40,9 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
private volatile Map<String, Integer> cache = new ConcurrentHashMap<>();
private final AtomicInteger hitCount = new AtomicInteger();
private final AtomicInteger missCount = new AtomicInteger();
+ // bumped by every eviction after a delete, before the key leaves the
cache: an add that read the row before the
+ // delete committed sees the change after caching the key, and removes it
again
+ private final AtomicLong evictions = new AtomicLong();
private String queryAllString = DEFAULT_QUERY_ALL_STRING;
public JdbcCachedMessageIdRepository() {
@@ -73,17 +77,18 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
@Override
public boolean add(final String key) {
- Integer previousValue = cache.getOrDefault(key, 0);
- if (previousValue != 0) {
- cache.merge(key, 1, Integer::sum);
+ // one atomic step: a key evicted meanwhile by a remove on another
thread is not put back
+ if (cache.computeIfPresent(key, (k, count) -> count + 1) != null) {
hitCount.incrementAndGet();
return false;
}
missCount.incrementAndGet();
+ // read before the row is: a remove or clear that deletes it after
this read changes the counter
+ long seen = evictions.get();
boolean added = super.add(key);
// only remember the key after the database insert succeeded,
otherwise a failed insert
// would leave the key cached and every redelivery would be rejected
as a duplicate
- cacheAfterCommit(key);
+ cacheAfterCommit(key, seen);
return added;
}
@@ -103,10 +108,15 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
cache.remove(key);
cancelCacheAfterCommit(key);
boolean removed = super.remove(key);
- // and again after the delete: an add of the key on another thread
whose transaction committed while this delete
- // waited for its row lock may have cached it meanwhile. This narrows
the window in which the cache keeps a
- // deleted key; it does not close it, as that add's afterCommit can
still run after this line.
- cache.remove(key);
+ if (removed) {
+ // and again after the delete: an add of the key on another thread
that read the row before the delete
+ // committed may have cached it meanwhile, or caches it later and
then sees the changed counter (see cacheKey)
+ evict(key);
+ // in a transaction of the caller the row is deleted when it
commits: until then another thread still sees
+ // the row and caches the key (its add is a duplicate), so the key
is removed from the cache again after the
+ // commit
+ evictAfterCommit(key);
+ }
return removed;
}
@@ -117,8 +127,9 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
hitCount.set(0);
missCount.set(0);
super.clear();
- // and again after the delete, as in remove
- cache.clear();
+ // and again after the delete, and after the commit of a transaction
of the caller, as in remove
+ evict(null);
+ evictAfterCommit(null);
}
/**
@@ -126,11 +137,41 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
* (for example a transacted route), the row is gone, and a cached key
would make the redelivery of the message a
* duplicate. Without a transaction with synchronization the key is cached
at once.
*/
- private void cacheAfterCommit(String key) {
+ private void cacheAfterCommit(String key, long seen) {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
- TransactionSynchronizationManager.registerSynchronization(new
CacheAfterCommit(this, key));
+ // a remove of the key earlier in the same transaction does not
evict it at the commit: the row is there again
+ cancelEvictAfterCommit(key);
+ TransactionSynchronizationManager.registerSynchronization(new
CacheAfterCommit(this, key, seen));
} else {
- cache.merge(key, 1, Integer::sum);
+ cacheKey(key, seen);
+ }
+ }
+
+ /**
+ * Caches the key, unless a remove or clear evicted keys since the add
read the row: that delete may have removed
+ * the row, and its eviction may have run before this merge. Either the
eviction runs after the merge and removes
+ * the key, or it has changed the counter before the check below, so a
read of the row before a delete committed
+ * does not leave its key in the cache (with read committed: a snapshot of
a transaction of the caller older than
+ * the counter is not covered). An add that runs alongside a remove or
clear of any key is therefore not cached,
+ * which costs one SELECT later.
+ */
+ private void cacheKey(String key, long seen) {
+ cache.merge(key, 1, Integer::sum);
+ if (evictions.get() != seen) {
+ cache.remove(key);
+ }
+ }
+
+ /**
+ * Removes the key (or every key when the key is null) from the cache
after its row was deleted. The counter changes
+ * first, so an add that cached the key meanwhile either sees the change
or is removed here.
+ */
+ private void evict(String key) {
+ evictions.incrementAndGet();
+ if (key == null) {
+ cache.clear();
+ } else {
+ cache.remove(key);
}
}
@@ -149,21 +190,68 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
}
}
+ /**
+ * Removes the key (or every key when the key is null) from the cache
again once the transaction of the caller that
+ * deleted its row commits. Without a transaction with synchronization the
delete has committed already. When the
+ * transaction of the caller is of another transaction manager (another
DataSource, JMS), the delete committed in a
+ * transaction of its own already, and this only costs a cache miss.
+ */
+ private void evictAfterCommit(String key) {
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
EvictAfterCommit(this, key));
+ }
+ }
+
+ /**
+ * An add of the key later in the same transaction cancels its eviction
after the commit, as the row is there again
+ * when the transaction commits. The eviction of a clear is kept: the add
is then not cached, which costs one
+ * SELECT.
+ */
+ private void cancelEvictAfterCommit(String key) {
+ for (TransactionSynchronization synchronization :
TransactionSynchronizationManager.getSynchronizations()) {
+ if (synchronization instanceof EvictAfterCommit pending &&
pending.repository == this
+ && key.equals(pending.key)) {
+ pending.cancelled = true;
+ }
+ }
+ }
+
+ private static final class EvictAfterCommit implements
TransactionSynchronization {
+ private final JdbcCachedMessageIdRepository repository;
+ private final String key;
+ // set and read by the thread of the transaction
+ private boolean cancelled;
+
+ private EvictAfterCommit(JdbcCachedMessageIdRepository repository,
String key) {
+ this.repository = repository;
+ this.key = key;
+ }
+
+ @Override
+ public void afterCommit() {
+ if (!cancelled) {
+ repository.evict(key);
+ }
+ }
+ }
+
private static final class CacheAfterCommit implements
TransactionSynchronization {
private final JdbcCachedMessageIdRepository repository;
private final String key;
+ private final long seen;
// set and read by the thread of the transaction
private boolean cancelled;
- private CacheAfterCommit(JdbcCachedMessageIdRepository repository,
String key) {
+ private CacheAfterCommit(JdbcCachedMessageIdRepository repository,
String key, long seen) {
this.repository = repository;
this.key = key;
+ this.seen = seen;
}
@Override
public void afterCommit() {
if (!cancelled) {
- repository.cache.merge(key, 1, Integer::sum);
+ repository.cacheKey(key, seen);
}
}
}
@@ -185,6 +273,8 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
}
public void reload() {
+ // a remove or clear while the keys are read may delete a row the
query saw (as for an add, see cacheKey)
+ long seen = evictions.get();
transactionTemplate.execute(status -> {
try {
cache = jdbcTemplate.query(getQueryAllString(), resultSet -> {
@@ -204,6 +294,10 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
return Boolean.TRUE;
});
+ if (evictions.get() != seen) {
+ // the loaded keys may include a deleted row: start empty instead
(only misses until they are cached again)
+ cache.clear();
+ }
}
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryTransactionTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryTransactionTest.java
index 656df26634b8..99914d274e53 100644
---
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryTransactionTest.java
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryTransactionTest.java
@@ -16,6 +16,12 @@
*/
package org.apache.camel.processor.idempotent.jdbc;
+import java.sql.Timestamp;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BooleanSupplier;
+
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -126,6 +132,166 @@ public class JdbcCachedMessageIdRepositoryTransactionTest
{
assertTrue(repository.add(KEY));
}
+ @Test
+ public void
removeInACallerTransactionEvictsTheKeyThatAnotherThreadCachedBeforeTheCommit()
throws Exception {
+ assertTrue(repository.add(KEY));
+ // the other thread adds the key again while the transaction that
deletes its row is open: it still sees the
+ // row, so its add is a duplicate, and it caches the key
+ assertTrue(removeInAnOpenTransactionWhile(() ->
assertFalse(repository.add(KEY)), false));
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY), "the key whose row was deleted
must not stay a duplicate in the cache");
+ assertTrue(repository.add(KEY));
+ assertEquals(1, rows());
+ }
+
+ @Test
+ public void
clearInACallerTransactionEvictsTheKeyThatAnotherThreadCachedBeforeTheCommit()
throws Exception {
+ assertTrue(repository.add(KEY));
+ assertTrue(removeInAnOpenTransactionWhile(() ->
assertFalse(repository.add(KEY)), true));
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY), "the key whose row was deleted
must not stay a duplicate in the cache");
+ assertTrue(repository.add(KEY));
+ }
+
+ @Test
+ public void removeInACallerTransactionThatRollsBackKeepsTheRow() {
+ assertTrue(repository.add(KEY));
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.remove(KEY));
+ status.setRollbackOnly();
+ });
+
+ assertEquals(1, rows());
+ assertTrue(repository.contains(KEY));
+ assertFalse(repository.add(KEY));
+ }
+
+ @Test
+ public void removeAndAddInOneCallerTransactionIsCached() {
+ assertTrue(repository.add(KEY));
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.remove(KEY));
+ assertTrue(repository.add(KEY));
+ });
+ int hits = repository.getHitCount();
+
+ assertEquals(1, rows());
+ assertTrue(repository.contains(KEY));
+ assertEquals(hits + 1, repository.getHitCount(), "the key added after
the remove must be cached at the commit");
+ }
+
+ @Test
+ public void addOnAnotherThreadThatReadTheRowBeforeARemoveIsNotCached()
throws Exception {
+ // the row is there, but not in the cache (such as a row of another
node), so the add reads it
+ insertRowNotCached();
+ int misses = repository.getMissCount();
+ // the other thread reads the row (its add is a duplicate), the remove
deletes the row and evicts the key, and
+ // only then does the other thread cache the key, when its transaction
commits
+ assertFalse(inAnOpenTransactionWhile(() -> repository.add(KEY), () ->
assertTrue(repository.remove(KEY))));
+ assertEquals(misses + 1, repository.getMissCount(), "the add must have
read the row");
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY), "the key whose row was deleted
must not stay a duplicate in the cache");
+ assertTrue(repository.add(KEY));
+ assertEquals(1, rows());
+ }
+
+ @Test
+ public void addOnAnotherThreadThatReadTheRowBeforeAClearIsNotCached()
throws Exception {
+ insertRowNotCached();
+ assertFalse(inAnOpenTransactionWhile(() -> repository.add(KEY),
repository::clear));
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY), "the key whose row was deleted
must not stay a duplicate in the cache");
+ assertTrue(repository.add(KEY));
+ }
+
+ @Test
+ public void addOnAnotherThreadIsCachedWhenNoRowIsRemovedMeanwhile() throws
Exception {
+ assertTrue(inAnOpenTransactionWhile(() -> repository.add(KEY), () ->
assertFalse(repository.remove("OTHER"))));
+ int hits = repository.getHitCount();
+
+ assertTrue(repository.contains(KEY));
+ assertEquals(hits + 1, repository.getHitCount(), "the committed key
must be answered from the cache");
+ }
+
+ @Test
+ public void clearAndAddInOneCallerTransactionKeepsTheRow() {
+ assertTrue(repository.add(KEY));
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ repository.clear();
+ assertTrue(repository.add(KEY));
+ });
+
+ assertEquals(1, rows());
+ assertTrue(repository.contains(KEY));
+ assertFalse(repository.add(KEY));
+ }
+
+ /**
+ * Removes the key (or clears) in a transaction of the caller on another
thread, runs the action on this thread
+ * while that transaction is open, and then lets it commit. Returns
whether the remove returned true.
+ */
+ private boolean removeInAnOpenTransactionWhile(Runnable action, boolean
clear) throws Exception {
+ return inAnOpenTransactionWhile(() -> {
+ if (clear) {
+ repository.clear();
+ return true;
+ }
+ return repository.remove(KEY);
+ }, action);
+ }
+
+ /**
+ * Runs the step in a transaction of the caller on another thread, runs
the action on this thread while that
+ * transaction is open, and then lets it commit. Returns the result of the
step.
+ */
+ private boolean inAnOpenTransactionWhile(BooleanSupplier step, Runnable
action) throws Exception {
+ CountDownLatch stepped = new CountDownLatch(1);
+ CountDownLatch commit = new CountDownLatch(1);
+ AtomicReference<Boolean> result = new AtomicReference<>();
+ AtomicReference<Throwable> failure = new AtomicReference<>();
+ Thread caller = new Thread(() -> {
+ try {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ result.set(step.getAsBoolean());
+ stepped.countDown();
+ try {
+ assertTrue(commit.await(20, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ });
+ } catch (Throwable t) {
+ failure.set(t);
+ } finally {
+ stepped.countDown();
+ }
+ }, "caller-transaction");
+ caller.start();
+ try {
+ assertTrue(stepped.await(20, TimeUnit.SECONDS));
+ action.run();
+ } finally {
+ commit.countDown();
+ caller.join(TimeUnit.SECONDS.toMillis(20));
+ }
+ assertFalse(caller.isAlive());
+ if (failure.get() != null) {
+ throw new AssertionError("the transaction of the caller failed",
failure.get());
+ }
+ return result.get();
+ }
+
+ private void insertRowNotCached() {
+ jdbcTemplate.update("INSERT INTO CAMEL_MESSAGEPROCESSED
(processorName, messageId, createdAt) VALUES (?, ?, ?)",
+ PROCESSOR_NAME, KEY, new
Timestamp(System.currentTimeMillis()));
+ assertEquals(1, rows());
+ }
+
private int rows() {
return jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
CAMEL_MESSAGEPROCESSED WHERE processorName = ?",
Integer.class, PROCESSOR_NAME);
