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 abbf701df316 CAMEL-26770: camel-sql - JdbcCachedMessageIdRepository:
cache a key only after its row is committed (#29533)
abbf701df316 is described below
commit abbf701df316e7de8467e2b79760ad4942dc3856
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Nov 10 18:22:33 2026 +0530
CAMEL-26770: camel-sql - JdbcCachedMessageIdRepository: cache a key only
after its row is committed (#29533)
JdbcCachedMessageIdRepository.add put the key in its in-memory cache as soon
as super.add returned, before the transaction that inserted the row
committed. When the add joined a caller's transaction that later rolled
back,
the cache kept a key whose row was never stored, and later duplicates of
that
message were rejected. Cache the key only after the row is committed, and
document the extra rollback of a second uncommitted add.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/sql-component.adoc | 9 ++
.../camel-sql/src/main/docs/sql-component.adoc | 9 ++
.../jdbc/JdbcCachedMessageIdRepository.java | 62 +++++++++-
...bcCachedMessageIdRepositoryTransactionTest.java | 133 +++++++++++++++++++++
4 files changed, 211 insertions(+), 2 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 b1823c53b43c..9ba79a81b105 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
@@ -775,6 +775,15 @@ checked first before passing through to the original
implementation.
As with all cache implementations, there are considerations that should
be made with regard to stale data and your specific usage.
+In a transaction, a key is added to the cache only once the transaction
+that inserted its row commits, so a rolled-back add is never cached. A
+second add of the same key on the same node, made while the first is still
+uncommitted, therefore does not hit the cache: it reaches the primary key,
+gets a duplicate key error, and its transaction is marked rollback-only, so
+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.
+
=== 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 b1823c53b43c..9ba79a81b105 100644
--- a/components/camel-sql/src/main/docs/sql-component.adoc
+++ b/components/camel-sql/src/main/docs/sql-component.adoc
@@ -775,6 +775,15 @@ checked first before passing through to the original
implementation.
As with all cache implementations, there are considerations that should
be made with regard to stale data and your specific usage.
+In a transaction, a key is added to the cache only once the transaction
+that inserted its row commits, so a rolled-back add is never cached. A
+second add of the same key on the same node, made while the first is still
+uncommitted, therefore does not hit the cache: it reaches the primary key,
+gets a duplicate key error, and its transaction is marked rollback-only, so
+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.
+
=== 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 07c2bbb91f42..546d446bbfc7 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
@@ -24,6 +24,8 @@ import javax.sql.DataSource;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.support.TransactionSynchronization;
+import
org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
/**
@@ -81,7 +83,7 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
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
- cache.merge(key, 1, Integer::sum);
+ cacheAfterCommit(key);
return added;
}
@@ -99,15 +101,71 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
@Override
public boolean remove(String key) {
cache.remove(key);
- return super.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);
+ return removed;
}
@Override
public void clear() {
cache.clear();
+ cancelCacheAfterCommit(null);
hitCount.set(0);
missCount.set(0);
super.clear();
+ // and again after the delete, as in remove
+ cache.clear();
+ }
+
+ /**
+ * Caches the key once the transaction that inserted its row commits. In a
transaction of the caller that rolls back
+ * (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) {
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ TransactionSynchronizationManager.registerSynchronization(new
CacheAfterCommit(this, key));
+ } else {
+ cache.merge(key, 1, Integer::sum);
+ }
+ }
+
+ /**
+ * A remove of the key (or a clear when the key is null) later in the same
transaction cancels its caching after the
+ * commit.
+ */
+ private void cancelCacheAfterCommit(String key) {
+ if (TransactionSynchronizationManager.isSynchronizationActive()) {
+ for (TransactionSynchronization synchronization :
TransactionSynchronizationManager.getSynchronizations()) {
+ if (synchronization instanceof CacheAfterCommit pending &&
pending.repository == this
+ && (key == null || pending.key.equals(key))) {
+ pending.cancelled = true;
+ }
+ }
+ }
+ }
+
+ private static final class CacheAfterCommit implements
TransactionSynchronization {
+ private final JdbcCachedMessageIdRepository repository;
+ private final String key;
+ // set and read by the thread of the transaction
+ private boolean cancelled;
+
+ private CacheAfterCommit(JdbcCachedMessageIdRepository repository,
String key) {
+ this.repository = repository;
+ this.key = key;
+ }
+
+ @Override
+ public void afterCommit() {
+ if (!cancelled) {
+ repository.cache.merge(key, 1, Integer::sum);
+ }
+ }
}
public String getQueryAllString() {
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
new file mode 100644
index 000000000000..656df26634b8
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryTransactionTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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 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 cached only once the transaction that inserted its row commits. In
a transaction of the caller that rolls
+ * back (for example a transacted route), the row is gone, so the redelivery
of the message must not be rejected as a
+ * duplicate from the cache.
+ */
+public class JdbcCachedMessageIdRepositoryTransactionTest {
+
+ private static final String PROCESSOR_NAME = "myProcessorName";
+ private static final String KEY = "KEY";
+
+ private EmbeddedDatabase dataSource;
+ private JdbcTemplate jdbcTemplate;
+ private JdbcCachedMessageIdRepository repository;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcTemplate = new JdbcTemplate(dataSource);
+ repository = new JdbcCachedMessageIdRepository(dataSource,
PROCESSOR_NAME);
+ repository.start();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ if (repository != null) {
+ repository.stop();
+ }
+ if (dataSource != null) {
+ dataSource.shutdown();
+ }
+ }
+
+ @Test
+ public void addInACallerTransactionThatRollsBackIsNotCached() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ status.setRollbackOnly();
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY), "the key of a rolled back add
must not be a duplicate");
+ // the redelivery of the message is processed
+ assertTrue(repository.add(KEY));
+ assertEquals(1, rows());
+ }
+
+ @Test
+ public void addInACallerTransactionIsCachedAfterTheCommit() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ // not cached before the commit: contains reads the row
+ int misses = repository.getMissCount();
+ assertTrue(repository.contains(KEY));
+ assertEquals(misses + 1, repository.getMissCount());
+ });
+ int hits = repository.getHitCount();
+
+ assertEquals(1, rows());
+ assertTrue(repository.contains(KEY));
+ assertEquals(hits + 1, repository.getHitCount(), "the committed key
must be answered from the cache");
+ }
+
+ @Test
+ public void addIsCached() {
+ assertTrue(repository.add(KEY));
+ int hits = repository.getHitCount();
+
+ assertFalse(repository.add(KEY));
+ assertEquals(hits + 1, repository.getHitCount());
+ }
+
+ @Test
+ public void addAndRemoveInOneCallerTransactionIsNotCached() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ assertTrue(repository.remove(KEY));
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY));
+ assertTrue(repository.add(KEY));
+ }
+
+ @Test
+ public void addAndClearInOneCallerTransactionIsNotCached() {
+ repository.transactionTemplate.executeWithoutResult(status -> {
+ assertTrue(repository.add(KEY));
+ repository.clear();
+ });
+
+ assertEquals(0, rows());
+ assertFalse(repository.contains(KEY));
+ assertTrue(repository.add(KEY));
+ }
+
+ private int rows() {
+ return jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
CAMEL_MESSAGEPROCESSED WHERE processorName = ?",
+ Integer.class, PROCESSOR_NAME);
+ }
+}
