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 d70d1b67fa7c CAMEL-26708: camel-sql - JdbcKeyValueRepository
putIfAbsent has one winner and expiry keeps a live entry (#29405)
d70d1b67fa7c is described below
commit d70d1b67fa7cbbc7dbf6752553168acd154b0331
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 16:31:10 2026 +0530
CAMEL-26708: camel-sql - JdbcKeyValueRepository putIfAbsent has one winner
and expiry keeps a live entry (#29405)
* CAMEL-26708: camel-sql - JdbcKeyValueRepository putIfAbsent has one
winner and expiry keeps a live entry
* CAMEL-26708: camel-sql - JdbcKeyValueRepository docs: class name, atomic
putIfAbsent, synchronised clocks
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../catalog/beans/JdbcKeyValueRepository.json | 2 +-
.../camel/catalog/docs/keyValueRepository.adoc | 21 +-
.../jdbc/JdbcKeyValueRepositoryConfigurer.java | 6 +
.../apache/camel/bean/JdbcKeyValueRepository.json | 2 +-
.../keyvalue/jdbc/JdbcKeyValueRepository.java | 74 ++++--
.../JdbcKeyValueRepositoryConcurrencyTest.java | 255 +++++++++++++++++++++
.../modules/eips/pages/keyValueRepository.adoc | 21 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 15 ++
.../dsl/jbang/core/commands/ai/eip-samples.json | 2 +-
9 files changed, 356 insertions(+), 42 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json
index 655fa16a61ca..7e84b41ecbe3 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/beans/JdbcKeyValueRepository.json
@@ -10,7 +10,7 @@
"groupId": "org.apache.camel",
"artifactId": "camel-sql",
"version": "4.24.0-SNAPSHOT",
- "properties": { "jdbcTemplate": { "index": 0, "kind": "property",
"displayName": "Jdbc Template", "required": true, "type": "object", "javaType":
"org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired":
false, "secret": false, "description": "The Spring JdbcTemplate to use for
connecting to the database" }, "transactionTemplate": { "index": 1, "kind":
"property", "displayName": "Transaction Template", "required": true, "type":
"object", "javaType": "org.springf [...]
+ "properties": { "jdbcTemplate": { "index": 0, "kind": "property",
"displayName": "Jdbc Template", "required": true, "type": "object", "javaType":
"org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired":
false, "secret": false, "description": "The Spring JdbcTemplate to use for
connecting to the database" }, "transactionTemplate": { "index": 1, "kind":
"property", "displayName": "Transaction Template", "required": true, "type":
"object", "javaType": "org.springf [...]
}
}
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
index 119785e05086..139f35c2945a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/keyValueRepository.adoc
@@ -74,7 +74,7 @@ Choosing the right backend is typically a matter of:
| `JdbcKeyValueRepository`
| `camel-sql`
| Any JDBC DataSource; portable; optional auto-DDL; ACID via DataSource
transactions
-| ✗ (non-atomic)
+| ✓ (primary key)
| ✗ (client-side)
| `JpaKeyValueRepository`
@@ -588,7 +588,7 @@ YAML DSL::
----
- beans:
- name: jdbcKvr
- type: org.apache.camel.component.sql.JdbcKeyValueRepository
+ type: org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository
properties:
dataSource: "#bean:myDataSource"
tableName: product_cache
@@ -596,10 +596,14 @@ YAML DSL::
----
====
-NOTE: `JdbcKeyValueRepository` does not implement atomic `putIfAbsent` —
concurrent inserts
-may produce duplicate-key exceptions that are caught and treated as a "key
already exists"
-condition. This is safe for idempotency at low concurrency, but for
high-throughput
-exactly-once semantics, prefer a backend with native CAS.
+NOTE: `JdbcKeyValueRepository.putIfAbsent` relies on the primary key of the
table: when concurrent
+inserts of the same key fail with a duplicate key, the transaction is rolled
back and the key is read
+again in a new transaction, which returns the value stored by the other
caller. When `putIfAbsent`
+runs in a transaction of the caller, the `DuplicateKeyException` is thrown
instead.
+
+An entry with a TTL expires lazily: the node that reads it compares its own
clock with the `EXPIRES_AT`
+value written by the node that stored it, so all nodes need synchronised
clocks. A node whose clock runs
+ahead treats the entry as expired, and deletes it, before its TTL has passed.
=== JpaKeyValueRepository
@@ -657,9 +661,8 @@ YAML DSL::
----
====
-NOTE: Like `JdbcKeyValueRepository`, the JPA backend does not provide atomic
-`putIfAbsent`. Idempotency under concurrent load relies on unique-constraint
violations
-being caught at the database level.
+NOTE: The JPA backend does not provide atomic `putIfAbsent`. Idempotency under
concurrent
+load relies on unique-constraint violations being caught at the database level.
=== CassandraKeyValueRepository
diff --git
a/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java
b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java
index 35e4baa2c619..9f0fe963cbe3 100644
---
a/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java
+++
b/components/camel-sql/src/generated/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConfigurer.java
@@ -32,6 +32,8 @@ public class JdbcKeyValueRepositoryConfigurer extends
org.apache.camel.support.c
case "createTableIfNotExists":
target.setCreateTableIfNotExists(property(camelContext, boolean.class, value));
return true;
case "deleteexpiredstring":
case "deleteExpiredString":
target.setDeleteExpiredString(property(camelContext, java.lang.String.class,
value)); return true;
+ case "deleteifexpiredstring":
+ case "deleteIfExpiredString":
target.setDeleteIfExpiredString(property(camelContext, java.lang.String.class,
value)); return true;
case "deleteifvaluestring":
case "deleteIfValueString":
target.setDeleteIfValueString(property(camelContext, java.lang.String.class,
value)); return true;
case "deletestring":
@@ -69,6 +71,8 @@ public class JdbcKeyValueRepositoryConfigurer extends
org.apache.camel.support.c
case "createTableIfNotExists": return boolean.class;
case "deleteexpiredstring":
case "deleteExpiredString": return java.lang.String.class;
+ case "deleteifexpiredstring":
+ case "deleteIfExpiredString": return java.lang.String.class;
case "deleteifvaluestring":
case "deleteIfValueString": return java.lang.String.class;
case "deletestring":
@@ -107,6 +111,8 @@ public class JdbcKeyValueRepositoryConfigurer extends
org.apache.camel.support.c
case "createTableIfNotExists": return
target.isCreateTableIfNotExists();
case "deleteexpiredstring":
case "deleteExpiredString": return target.getDeleteExpiredString();
+ case "deleteifexpiredstring":
+ case "deleteIfExpiredString": return target.getDeleteIfExpiredString();
case "deleteifvaluestring":
case "deleteIfValueString": return target.getDeleteIfValueString();
case "deletestring":
diff --git
a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json
b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json
index 655fa16a61ca..7e84b41ecbe3 100644
---
a/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json
+++
b/components/camel-sql/src/generated/resources/META-INF/services/org/apache/camel/bean/JdbcKeyValueRepository.json
@@ -10,7 +10,7 @@
"groupId": "org.apache.camel",
"artifactId": "camel-sql",
"version": "4.24.0-SNAPSHOT",
- "properties": { "jdbcTemplate": { "index": 0, "kind": "property",
"displayName": "Jdbc Template", "required": true, "type": "object", "javaType":
"org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired":
false, "secret": false, "description": "The Spring JdbcTemplate to use for
connecting to the database" }, "transactionTemplate": { "index": 1, "kind":
"property", "displayName": "Transaction Template", "required": true, "type":
"object", "javaType": "org.springf [...]
+ "properties": { "jdbcTemplate": { "index": 0, "kind": "property",
"displayName": "Jdbc Template", "required": true, "type": "object", "javaType":
"org.springframework.jdbc.core.JdbcTemplate", "deprecated": false, "autowired":
false, "secret": false, "description": "The Spring JdbcTemplate to use for
connecting to the database" }, "transactionTemplate": { "index": 1, "kind":
"property", "displayName": "Transaction Template", "required": true, "type":
"object", "javaType": "org.springf [...]
}
}
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
index f05b43b16a2d..383a1c161b91 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepository.java
@@ -79,6 +79,11 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
= "UPDATE CAMEL_KEYVALUE SET ITEM_VALUE = ?, EXPIRES_AT = ? WHERE
ITEM_KEY = ? AND ITEM_VALUE = ?";
protected static final String DEFAULT_DELETE_IF_VALUE_STRING
= "DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ? AND ITEM_VALUE =
?";
+ protected static final String DEFAULT_DELETE_IF_EXPIRED_STRING
+ = "DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ? AND EXPIRES_AT >
0 AND EXPIRES_AT <= ?";
+
+ // how many transactions putIfAbsent may use when concurrent inserts of
the same key keep failing
+ private static final int PUT_IF_ABSENT_ATTEMPTS = 3;
private static final Logger LOG =
LoggerFactory.getLogger(JdbcKeyValueRepository.class);
@@ -113,6 +118,8 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
private String updateIfValueString = DEFAULT_UPDATE_IF_VALUE_STRING;
@Metadata(label = "advanced", description = "SQL query to use for
conditional delete (CAS delete)")
private String deleteIfValueString = DEFAULT_DELETE_IF_VALUE_STRING;
+ @Metadata(label = "advanced", description = "SQL query to use for deleting
an entry by key only if it has expired")
+ private String deleteIfExpiredString = DEFAULT_DELETE_IF_EXPIRED_STRING;
@Metadata(label = "advanced,security",
description = "Sets an ObjectInputFilter pattern
(jdk.serialFilter syntax) applied when deserializing"
+ " values read back from the repository. When not
set, the JVM-wide jdk.serialFilter is"
@@ -186,6 +193,7 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
deleteExpiredString =
DEFAULT_DELETE_EXPIRED_STRING.replace(DEFAULT_TABLENAME, tableName);
updateIfValueString =
DEFAULT_UPDATE_IF_VALUE_STRING.replace(DEFAULT_TABLENAME, tableName);
deleteIfValueString =
DEFAULT_DELETE_IF_VALUE_STRING.replace(DEFAULT_TABLENAME, tableName);
+ deleteIfExpiredString =
DEFAULT_DELETE_IF_EXPIRED_STRING.replace(DEFAULT_TABLENAME, tableName);
}
}
@@ -282,28 +290,41 @@ public class JdbcKeyValueRepository extends
ServiceSupport implements KeyValueRe
transactionTemplate.executeWithoutResult(status ->
jdbcTemplate.update(getClearString()));
}
+ /**
+ * Stores the value only if no non-expired entry exists for the key. The
primary key of the table decides between
+ * concurrent callers: when the INSERT fails with a duplicate key, the
transaction is rolled back and the key is
+ * read again in a new transaction, which returns the value stored by the
other caller, or inserts again if that
+ * entry has been deleted meanwhile.
+ *
+ * @throws DuplicateKeyException if the INSERT failed in all attempts, or
at once in a transaction of the caller
+ */
@Override
public Object putIfAbsent(String key, Object value, Duration ttl) {
- return transactionTemplate.execute(status -> {
- // check if a non-expired entry already exists
- Object existing = doGet(key);
- if (existing != null) {
- return existing;
- }
- // attempt to insert
- long expiresAt = toExpiresAt(ttl);
+ byte[] bytes = KeyValueRepositoryHelper.serialize(value);
+ boolean[] joined = { false };
+ for (int attempt = 1;; attempt++) {
try {
- jdbcTemplate.update(getInsertString(), key,
KeyValueRepositoryHelper.serialize(value), expiresAt);
- return null;
+ return transactionTemplate.execute(status -> {
+ joined[0] = status.hasTransaction() &&
!status.isNewTransaction();
+ // check if a non-expired entry already exists
+ Object existing = doGet(key);
+ if (existing != null) {
+ return existing;
+ }
+ // attempt to insert
+ long expiresAt = toExpiresAt(ttl);
+ jdbcTemplate.update(getInsertString(), key, bytes,
expiresAt);
+ return null;
+ });
} catch (DuplicateKeyException e) {
- // concurrent insert race -- another thread/node won
- LOG.debug("Concurrent insert race for key '{}' -- another
thread won, treating as existing", key);
- status.setRollbackOnly();
- // re-read the value that the other thread inserted
- Object concurrentValue = doGet(key);
- return concurrentValue;
+ // another thread/node inserted the key: read it again in a
new transaction, as the failed one may
+ // not see that entry (REPEATABLE READ) or the entry may be
deleted meanwhile
+ if (joined[0] || attempt >= PUT_IF_ABSENT_ATTEMPTS) {
+ throw e;
+ }
+ LOG.debug("Concurrent insert race for key '{}' -- reading it
again in a new transaction", key);
}
- });
+ }
}
/**
@@ -361,29 +382,32 @@ public class JdbcKeyValueRepository extends
ServiceSupport implements KeyValueRe
/**
* Internal get that reads the value for the given key within the current
transaction. If the entry has expired, it
- * is deleted and {@code null} is returned.
+ * is deleted, unless it has been stored again meanwhile, and {@code null}
is returned.
*
* @param key the key to look up
* @return the deserialized value, or {@code null} if not found or
expired
*/
private Object doGet(String key) {
try {
- // Use a two-element array as a container: [0] = deserialized
value, [1] = expired flag.
+ // Use a single-element array as the expired flag.
// The DELETE for expired rows is performed after queryForObject
returns so that we
// do not execute a statement while the ResultSet for the SELECT
is still open (some
// JDBC drivers dislike that).
final boolean[] expired = { false };
+ final long now = System.currentTimeMillis();
Object value = jdbcTemplate.queryForObject(getSelectString(), (rs,
rowNum) -> {
byte[] bytes = rs.getBytes(1);
long expiresAt = rs.getLong(2);
- if (expiresAt > 0 && System.currentTimeMillis() >= expiresAt) {
+ if (expiresAt > 0 && now >= expiresAt) {
expired[0] = true;
return null;
}
return KeyValueRepositoryHelper.deserialize(bytes,
deserializationFilter);
}, key);
if (expired[0]) {
- jdbcTemplate.update(getDeleteString(), key);
+ // delete the row only if it is still expired: another
transaction may have stored the key again
+ // since it was read, and that entry must be kept
+ jdbcTemplate.update(getDeleteIfExpiredString(), key, now);
return null;
}
return value;
@@ -526,6 +550,14 @@ public class JdbcKeyValueRepository extends ServiceSupport
implements KeyValueRe
this.deleteIfValueString = deleteIfValueString;
}
+ public String getDeleteIfExpiredString() {
+ return deleteIfExpiredString;
+ }
+
+ public void setDeleteIfExpiredString(String deleteIfExpiredString) {
+ this.deleteIfExpiredString = deleteIfExpiredString;
+ }
+
public String getDeserializationFilter() {
return deserializationFilter;
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConcurrencyTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConcurrencyTest.java
new file mode 100644
index 000000000000..3486981997fb
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/keyvalue/jdbc/JdbcKeyValueRepositoryConcurrencyTest.java
@@ -0,0 +1,255 @@
+/*
+ * 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.keyvalue.jdbc;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.sql.DataSource;
+
+import org.apache.camel.support.KeyValueRepositoryHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Two repositories (two nodes) on one database. The statements of node A can
be paused, so that the calls of both nodes
+ * interleave in a given order.
+ */
+class JdbcKeyValueRepositoryConcurrencyTest {
+
+ private EmbeddedDatabase dataSource;
+ private PausingJdbcTemplate jdbcA;
+ private JdbcKeyValueRepository nodeA;
+ private JdbcKeyValueRepository nodeB;
+ private final ExecutorService executor =
Executors.newSingleThreadExecutor();
+
+ private void startNodes(int isolationLevel) throws Exception {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcA = new PausingJdbcTemplate(dataSource);
+ nodeA = new JdbcKeyValueRepository(jdbcA,
transactionTemplate(isolationLevel));
+ nodeB = new JdbcKeyValueRepository(new JdbcTemplate(dataSource),
transactionTemplate(isolationLevel));
+ nodeA.init();
+ nodeA.start();
+ nodeB.init();
+ nodeB.start();
+ }
+
+ private TransactionTemplate transactionTemplate(int isolationLevel) {
+ TransactionTemplate template = new TransactionTemplate(new
DataSourceTransactionManager(dataSource));
+ template.setIsolationLevel(isolationLevel);
+ return template;
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (jdbcA != null) {
+ jdbcA.releaseAll();
+ }
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ if (nodeA != null) {
+ nodeA.stop();
+ nodeB.stop();
+ }
+ if (dataSource != null) {
+ dataSource.shutdown();
+ }
+ }
+
+ @Test
+ void putIfAbsentHasOneWinnerUnderRepeatableRead() throws Exception {
+ startNodes(TransactionDefinition.ISOLATION_REPEATABLE_READ);
+ Gate beforeInsert = jdbcA.pause("INSERT", 1);
+
+ Future<Object> a = executor.submit(() -> nodeA.putIfAbsent("order-1",
"A", null));
+ // A has read the key (absent), so its snapshot does not show what B
stores now
+ beforeInsert.awaitReached();
+ assertNull(nodeB.putIfAbsent("order-1", "B", null), "B stores the key
first");
+ // the INSERT of A fails with a duplicate key
+ beforeInsert.release();
+
+ assertEquals("B", a.get(10, TimeUnit.SECONDS), "A lost the race and
must get the value of B");
+ assertEquals("B", nodeA.get("order-1"));
+ }
+
+ @Test
+ void putIfAbsentDoesNotSucceedWithoutStoringTheValue() throws Exception {
+ startNodes(TransactionDefinition.ISOLATION_READ_COMMITTED);
+ Gate beforeInsert = jdbcA.pause("INSERT", 1);
+ Gate beforeSecondRead = jdbcA.pause("SELECT", 2);
+
+ Future<Object> a = executor.submit(() -> nodeA.putIfAbsent("lease-1",
"A", null));
+ beforeInsert.awaitReached();
+ assertNull(nodeB.putIfAbsent("lease-1", "B", null), "B stores the key
first");
+ // the INSERT of A fails with a duplicate key, then A reads the key
again
+ beforeInsert.release();
+ beforeSecondRead.awaitReached();
+ assertEquals("B", nodeB.delete("lease-1"), "B releases the key");
+ beforeSecondRead.release();
+
+ // the key was free when A read it again, so A may get it, but then
its value must be stored
+ assertNull(a.get(10, TimeUnit.SECONDS));
+ assertEquals("A", nodeB.get("lease-1"), "A was told that its put
succeeded");
+ }
+
+ @Test
+ void readingAnExpiredEntryDoesNotDeleteTheEntryStoredMeanwhile() throws
Exception {
+ startNodes(TransactionDefinition.ISOLATION_READ_COMMITTED);
+ // an entry that expired long ago
+ nodeB.getJdbcTemplate().update(nodeB.getInsertString(), "job-42",
KeyValueRepositoryHelper.serialize("old"), 1L);
+ Gate beforeDelete = jdbcA.pause("DELETE", 1);
+
+ Future<Object> a = executor.submit(() -> nodeA.get("job-42"));
+ // A has read the entry as expired and is about to delete it
+ beforeDelete.awaitReached();
+ assertNull(nodeB.putIfAbsent("job-42", "B", null), "the expired entry
is absent, so B gets the key");
+ beforeDelete.release();
+
+ assertNull(a.get(10, TimeUnit.SECONDS), "A read the entry while it was
expired");
+ assertEquals("B", nodeB.get("job-42"), "the entry stored by B must be
kept");
+ assertEquals("B", nodeA.putIfAbsent("job-42", "A", null), "B still
holds the key");
+ }
+
+ @Test
+ void putIfAbsentGivesUpAfterThreeDuplicateKeys() throws Exception {
+ startNodes(TransactionDefinition.ISOLATION_READ_COMMITTED);
+ assertNull(nodeB.putIfAbsent("job-1", "B", null));
+ // node A does not find the stored entry, so each of its INSERTs fails
with a duplicate key
+ nodeA.setSelectString(nodeA.getSelectString() + " AND 1 = 0");
+
+ assertThrows(DuplicateKeyException.class, () ->
nodeA.putIfAbsent("job-1", "A", null));
+ assertEquals(3, jdbcA.statements("INSERT"));
+ }
+
+ @Test
+ void putIfAbsentInTransactionOfCallerDoesNotTryAgain() throws Exception {
+ startNodes(TransactionDefinition.ISOLATION_READ_COMMITTED);
+ assertNull(nodeB.putIfAbsent("job-1", "B", null));
+ nodeA.setSelectString(nodeA.getSelectString() + " AND 1 = 0");
+ TransactionTemplate caller =
transactionTemplate(TransactionDefinition.ISOLATION_READ_COMMITTED);
+
+ // the transaction of the caller cannot go on after the duplicate key,
so the key is not read again in it
+ assertThrows(DuplicateKeyException.class,
+ () -> caller.executeWithoutResult(status ->
nodeA.putIfAbsent("job-1", "A", null)));
+ assertEquals(1, jdbcA.statements("INSERT"));
+ }
+
+ /**
+ * Pauses the n-th statement of a {@link PausingJdbcTemplate} that starts
with the given SQL verb, until released.
+ */
+ private static final class Gate {
+ private final String verb;
+ private final int occurrence;
+ private final AtomicInteger seen = new AtomicInteger();
+ private final CountDownLatch reached = new CountDownLatch(1);
+ private final CountDownLatch released = new CountDownLatch(1);
+
+ private Gate(String verb, int occurrence) {
+ this.verb = verb;
+ this.occurrence = occurrence;
+ }
+
+ private void beforeStatement(String sql) {
+ if (sql.startsWith(verb) && seen.incrementAndGet() == occurrence) {
+ reached.countDown();
+ await(released);
+ }
+ }
+
+ void awaitReached() {
+ await(reached);
+ }
+
+ void release() {
+ released.countDown();
+ }
+
+ private static void await(CountDownLatch latch) {
+ try {
+ if (!latch.await(10, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Timeout waiting for the
other node");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+ }
+
+ private static final class PausingJdbcTemplate extends JdbcTemplate {
+ private final List<Gate> gates = new CopyOnWriteArrayList<>();
+ private final List<String> executed = new CopyOnWriteArrayList<>();
+
+ private PausingJdbcTemplate(DataSource dataSource) {
+ super(dataSource);
+ }
+
+ Gate pause(String verb, int occurrence) {
+ Gate gate = new Gate(verb, occurrence);
+ gates.add(gate);
+ return gate;
+ }
+
+ long statements(String verb) {
+ return executed.stream().filter(sql ->
sql.startsWith(verb)).count();
+ }
+
+ void releaseAll() {
+ gates.forEach(Gate::release);
+ }
+
+ private void beforeStatement(String sql) {
+ executed.add(sql);
+ gates.forEach(gate -> gate.beforeStatement(sql));
+ }
+
+ @Override
+ public <T> T queryForObject(String sql, RowMapper<T> rowMapper,
Object... args) {
+ beforeStatement(sql);
+ return super.queryForObject(sql, rowMapper, args);
+ }
+
+ @Override
+ public int update(String sql, Object... args) {
+ beforeStatement(sql);
+ return super.update(sql, args);
+ }
+ }
+}
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
index 119785e05086..139f35c2945a 100644
---
a/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
+++
b/core/camel-core-engine/src/main/docs/modules/eips/pages/keyValueRepository.adoc
@@ -74,7 +74,7 @@ Choosing the right backend is typically a matter of:
| `JdbcKeyValueRepository`
| `camel-sql`
| Any JDBC DataSource; portable; optional auto-DDL; ACID via DataSource
transactions
-| ✗ (non-atomic)
+| ✓ (primary key)
| ✗ (client-side)
| `JpaKeyValueRepository`
@@ -588,7 +588,7 @@ YAML DSL::
----
- beans:
- name: jdbcKvr
- type: org.apache.camel.component.sql.JdbcKeyValueRepository
+ type: org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository
properties:
dataSource: "#bean:myDataSource"
tableName: product_cache
@@ -596,10 +596,14 @@ YAML DSL::
----
====
-NOTE: `JdbcKeyValueRepository` does not implement atomic `putIfAbsent` —
concurrent inserts
-may produce duplicate-key exceptions that are caught and treated as a "key
already exists"
-condition. This is safe for idempotency at low concurrency, but for
high-throughput
-exactly-once semantics, prefer a backend with native CAS.
+NOTE: `JdbcKeyValueRepository.putIfAbsent` relies on the primary key of the
table: when concurrent
+inserts of the same key fail with a duplicate key, the transaction is rolled
back and the key is read
+again in a new transaction, which returns the value stored by the other
caller. When `putIfAbsent`
+runs in a transaction of the caller, the `DuplicateKeyException` is thrown
instead.
+
+An entry with a TTL expires lazily: the node that reads it compares its own
clock with the `EXPIRES_AT`
+value written by the node that stored it, so all nodes need synchronised
clocks. A node whose clock runs
+ahead treats the entry as expired, and deletes it, before its TTL has passed.
=== JpaKeyValueRepository
@@ -657,9 +661,8 @@ YAML DSL::
----
====
-NOTE: Like `JdbcKeyValueRepository`, the JPA backend does not provide atomic
-`putIfAbsent`. Idempotency under concurrent load relies on unique-constraint
violations
-being caught at the database level.
+NOTE: The JPA backend does not provide atomic `putIfAbsent`. Idempotency under
concurrent
+load relies on unique-constraint violations being caught at the database level.
=== CassandraKeyValueRepository
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 10ab4517e865..790e3865ee8b 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -2865,6 +2865,21 @@ A single `Map` body is now one batch item, and so is a
single `Map` in the `Came
(sql-stored). Before, its entries were used as the batch items and the call
failed (sql: no value found for the named
parameters, sql-stored: an `SQLException`).
+=== camel-sql - JdbcKeyValueRepository: deleteIfExpiredString, and putIfAbsent
after a duplicate key
+
+When `JdbcKeyValueRepository` reads an entry that has expired, it deletes the
row with the new `deleteIfExpiredString`
+statement (`DELETE FROM CAMEL_KEYVALUE WHERE ITEM_KEY = ? AND EXPIRES_AT > 0
AND EXPIRES_AT <= ?`) instead of
+`deleteString`, so that an entry another node stored for the key in the
meantime is kept. If you set `selectString`,
+`insertString` and `deleteString` for a table with other column names, set
`deleteIfExpiredString` as well; otherwise
+reading an expired entry fails with an SQL error. With `tableName` it is
derived from the default statement, as the
+other statements are.
+
+When the INSERT of `putIfAbsent` fails with a duplicate key, the key is read
again in a new transaction, which returns
+the value stored by the other caller (before, the key was read again in the
failed transaction, which could return
+`null` although nothing was stored). `putIfAbsent` now throws the
`DuplicateKeyException` when the INSERT fails three
+times, or at once when it runs in a transaction of the caller, as that
transaction cannot be used to read the key
+again.
+
=== camel-jta - an exchange marked rollback only last rolls back its
transaction
In a `transacted()` route that uses a JTA transaction manager,
`markRollbackOnlyLast()` now rolls back the
diff --git
a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/org/apache/camel/dsl/jbang/core/commands/ai/eip-samples.json
b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/org/apache/camel/dsl/jbang/core/commands/ai/eip-samples.json
index a4449af9330c..58c73ca7a719 100644
---
a/dsl/camel-jbang/camel-jbang-core/src/generated/resources/org/apache/camel/dsl/jbang/core/commands/ai/eip-samples.json
+++
b/dsl/camel-jbang/camel-jbang-core/src/generated/resources/org/apache/camel/dsl/jbang/core/commands/ai/eip-samples.json
@@ -590,7 +590,7 @@
},
{
"source": "keyValueRepository.adoc",
- "yaml": "- beans:\n - name: jdbcKvr\n type:
org.apache.camel.component.sql.JdbcKeyValueRepository\n properties:\n
dataSource: \"#bean:myDataSource\"\n tableName: product_cache\n
createTableIfNotExists: true\n"
+ "yaml": "- beans:\n - name: jdbcKvr\n type:
org.apache.camel.processor.keyvalue.jdbc.JdbcKeyValueRepository\n
properties:\n dataSource: \"#bean:myDataSource\"\n tableName:
product_cache\n createTableIfNotExists: true\n"
},
{
"source": "keyValueRepository.adoc",
