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 3c2e37175a54 CAMEL-26736: camel-jpa - JpaKeyValueRepository
putIfAbsent has one winner and expiry keeps a live entry (#29440)
3c2e37175a54 is described below
commit 3c2e37175a549ee4af58199a0d301911b515fced
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Nov 9 17:45:55 2026 +0530
CAMEL-26736: camel-jpa - JpaKeyValueRepository putIfAbsent has one winner
and expiry keeps a live entry (#29440)
* CAMEL-26736: camel-jpa - JpaKeyValueRepository putIfAbsent has one wi…
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/keyValueRepository.adoc | 9 +-
.../keyvalue/jpa/JpaKeyValueRepository.java | 98 +++---
.../jpa/JpaKeyValueRepositoryConcurrencyTest.java | 355 +++++++++++++++++++++
.../modules/eips/pages/keyValueRepository.adoc | 9 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 +
5 files changed, 440 insertions(+), 42 deletions(-)
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 750a45c396ba..b7e26662cd75 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
@@ -80,7 +80,7 @@ Choosing the right backend is typically a matter of:
| `JpaKeyValueRepository`
| `camel-jpa`
| JPA entity-backed; works with any JPA 2 provider (Hibernate, EclipseLink…)
-| ✗ (non-atomic)
+| ✓ (primary key)
| ✗ (client-side)
| `CassandraKeyValueRepository`
@@ -675,8 +675,11 @@ YAML DSL::
----
====
-NOTE: The JPA backend does not provide atomic `putIfAbsent`. Idempotency under
concurrent
-load relies on unique-constraint violations being caught at the database level.
+NOTE: `JpaKeyValueRepository.putIfAbsent` relies on the primary key of the
table: when concurrent
+inserts of the same key fail with a constraint violation, the transaction is
rolled back and the key
+is read again in a new transaction, which returns the value stored by the
other caller. After three
+failed inserts the `PersistenceException` is thrown. When `putIfAbsent` runs
in a transaction of the
+caller, the failed insert marks that transaction for rollback, so the caller's
transaction fails.
=== CassandraKeyValueRepository
diff --git
a/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
index 581f57101d3c..882321343ee2 100644
---
a/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
+++
b/components/camel-jpa/src/main/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepository.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.processor.keyvalue.jpa;
+import java.sql.SQLException;
import java.sql.SQLIntegrityConstraintViolationException;
import java.time.Duration;
import java.util.Iterator;
@@ -70,6 +71,12 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
= "select e from " + KeyValueEntry.class.getName() + " e";
private static final String QUERY_COUNT_VALID
= "select count(e) from " + KeyValueEntry.class.getName() + " e
where e.expiresAt = 0 or e.expiresAt > ?1";
+ private static final String DELETE_IF_EXPIRED
+ = "delete from " + KeyValueEntry.class.getName()
+ + " e where e.itemKey = ?1 and e.expiresAt > 0 and e.expiresAt
<= ?2";
+
+ // 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 String SOMETHING_WENT_WRONG
= "Something went wrong in JpaKeyValueRepository: %s";
@@ -126,8 +133,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
if (entry == null) {
rc[0] = null;
} else if (entry.isExpired()) {
- entityManager.remove(entry);
- entityManager.flush();
+ deleteIfExpired(entityManager, key);
rc[0] = null;
} else {
rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue(),
deserializationFilter);
@@ -200,8 +206,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
if (entry == null) {
rc[0] = null;
} else if (entry.isExpired()) {
- entityManager.remove(entry);
- entityManager.flush();
+ deleteIfExpired(entityManager, key);
rc[0] = null;
} else {
rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue(),
deserializationFilter);
@@ -236,8 +241,7 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
if (entry == null) {
rc[0] = Boolean.FALSE;
} else if (entry.isExpired()) {
- entityManager.remove(entry);
- entityManager.flush();
+ deleteIfExpired(entityManager, key);
rc[0] = Boolean.FALSE;
} else {
rc[0] = Boolean.TRUE;
@@ -315,8 +319,35 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
LOG.debug("clear the store {}", KeyValueEntry.class.getName());
}
+ /**
+ * 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 constraint violation,
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. In a transaction of the caller, the
failed insert marks that transaction for
+ * rollback.
+ *
+ * @throws PersistenceException if the insert failed in all attempts
+ */
@Override
public Object putIfAbsent(String key, Object value, Duration ttl) {
+ final byte[] serializedValue =
KeyValueRepositoryHelper.serialize(value);
+ for (int attempt = 1;; attempt++) {
+ try {
+ Object answer = doPutIfAbsent(key, serializedValue, ttl);
+ LOG.debug("putIfAbsent {} -> {}", key, answer != null ?
"existing" : "inserted");
+ return answer;
+ } catch (PersistenceException e) {
+ // 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 (attempt >= PUT_IF_ABSENT_ATTEMPTS ||
!isConstraintViolation(e)) {
+ throw e;
+ }
+ LOG.debug("Concurrent insert of key '{}' -- reading it again
in a new transaction", key);
+ }
+ }
+ }
+
+ private Object doPutIfAbsent(String key, byte[] serializedValue, Duration
ttl) {
final Object[] rc = new Object[1];
final EntityManager entityManager
= getTargetEntityManager(null, entityManagerFactory, false,
sharedEntityManager, true);
@@ -326,50 +357,29 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
entityManager.joinTransaction();
}
try {
- long expiresAt = toExpiresAt(ttl);
- byte[] serializedValue =
KeyValueRepositoryHelper.serialize(value);
-
KeyValueEntry entry = findByKey(entityManager, key);
if (entry != null && !entry.isExpired()) {
// key exists and is valid -- return existing value
rc[0] =
KeyValueRepositoryHelper.deserialize(entry.getItemValue(),
deserializationFilter);
- } else if (entry != null) {
- // key exists but expired -- update in place
- entry.setItemValue(serializedValue);
- entry.setExpiresAt(expiresAt);
- entityManager.merge(entry);
- entityManager.flush();
- rc[0] = null;
} else {
- // no entry -- persist new one
- entry = new KeyValueEntry(key, serializedValue, expiresAt);
- entityManager.persist(entry);
+ if (entry != null) {
+ // key exists but expired -- delete it, unless it has
been stored again meanwhile, in which
+ // case the insert below fails
+ deleteIfExpired(entityManager, key);
+ entityManager.detach(entry);
+ }
+ entityManager.persist(new KeyValueEntry(key,
serializedValue, toExpiresAt(ttl)));
entityManager.flush();
rc[0] = null;
}
} catch (Exception ex) {
- if (isConstraintViolation(ex)) {
- // concurrent insert of the same key -- treat as "already
present"
- LOG.debug("Concurrent insert detected for key: {}", key);
- // re-read to return the existing value
- try {
- KeyValueEntry existing = findByKey(entityManager, key);
- rc[0] = existing != null
- ?
KeyValueRepositoryHelper.deserialize(existing.getItemValue(),
deserializationFilter) : null;
- } catch (Exception inner) {
- // fall through with null
- rc[0] = null;
- }
- } else {
- String contextInfo = String.format(SOMETHING_WENT_WRONG,
ex.getMessage());
- throw new PersistenceException(contextInfo, ex);
- }
+ String contextInfo = String.format(SOMETHING_WENT_WRONG,
ex.getMessage());
+ throw new PersistenceException(contextInfo, ex);
} finally {
closeEntityManager(entityManager);
}
});
- LOG.debug("putIfAbsent {} -> {}", key, rc[0] != null ? "existing" :
"inserted");
return rc[0];
}
@@ -503,6 +513,17 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
return (KeyValueEntry) list.get(0);
}
+ /**
+ * Deletes the entry of the key only if it is still expired: another
transaction may have stored the key again since
+ * it was read, and that entry must be kept.
+ */
+ private static void deleteIfExpired(EntityManager entityManager, String
key) {
+ Query query = entityManager.createQuery(DELETE_IF_EXPIRED);
+ query.setParameter(1, key);
+ query.setParameter(2, System.currentTimeMillis());
+ query.executeUpdate();
+ }
+
private static long toExpiresAt(Duration ttl) {
if (ttl == null || ttl.isZero() || ttl.isNegative()) {
return 0;
@@ -529,6 +550,11 @@ public class JpaKeyValueRepository extends ServiceSupport
implements KeyValueRep
if (cause instanceof EntityExistsException) {
return true;
}
+ // drivers such as the PostgreSQL and SQL Server ones report it
only with the SQL state class 23
+ if (cause instanceof SQLException sqlException &&
sqlException.getSQLState() != null
+ && sqlException.getSQLState().startsWith("23")) {
+ return true;
+ }
cause = cause.getCause();
}
return false;
diff --git
a/components/camel-jpa/src/test/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConcurrencyTest.java
b/components/camel-jpa/src/test/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConcurrencyTest.java
new file mode 100644
index 000000000000..64401004e29c
--- /dev/null
+++
b/components/camel-jpa/src/test/java/org/apache/camel/processor/keyvalue/jpa/JpaKeyValueRepositoryConcurrencyTest.java
@@ -0,0 +1,355 @@
+/*
+ * 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.jpa;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+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 jakarta.persistence.EntityManager;
+import jakarta.persistence.EntityManagerFactory;
+import jakarta.persistence.Persistence;
+import jakarta.persistence.PersistenceException;
+import jakarta.persistence.Query;
+
+import org.apache.camel.component.jpa.DefaultTransactionStrategy;
+import org.apache.camel.support.KeyValueRepositoryHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+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 entity manager operations
of node A can be paused, so that the
+ * calls of both nodes interleave in a given order.
+ */
+class JpaKeyValueRepositoryConcurrencyTest {
+
+ private EntityManagerFactory entityManagerFactory;
+ private PausingEntityManagerFactory pausingA;
+ private JpaKeyValueRepository nodeA;
+ private JpaKeyValueRepository nodeB;
+ private final ExecutorService executor =
Executors.newSingleThreadExecutor();
+
+ @BeforeEach
+ void startNodes() throws Exception {
+ entityManagerFactory =
Persistence.createEntityManagerFactory("keyvalueDb",
+ Map.of("openjpa.ConnectionURL", "jdbc:h2:mem:kv-" +
UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"));
+ pausingA = new PausingEntityManagerFactory(entityManagerFactory);
+ nodeA = createNode(pausingA.proxy());
+ nodeB = createNode(entityManagerFactory);
+ }
+
+ private static JpaKeyValueRepository createNode(EntityManagerFactory emf)
throws Exception {
+ JpaKeyValueRepository node = new JpaKeyValueRepository(emf);
+ node.setTransactionStrategy(new DefaultTransactionStrategy(null, emf));
+ node.setJoinTransaction(false);
+ node.setSharedEntityManager(true);
+ node.init();
+ node.start();
+ return node;
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (pausingA != null) {
+ pausingA.releaseAll();
+ }
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS));
+ if (nodeA != null) {
+ nodeA.stop();
+ nodeB.stop();
+ }
+ if (entityManagerFactory != null) {
+ entityManagerFactory.close();
+ }
+ }
+
+ @Test
+ void putIfAbsentReturnsTheValueOfTheWinner() throws Exception {
+ Gate beforeInsert = pausingA.pause(2);
+
+ Future<Object> a = executor.submit(() -> nodeA.putIfAbsent("order-1",
"A", null));
+ // A has read the key (absent) and is about to insert it
+ beforeInsert.awaitReached();
+ assertNull(nodeB.putIfAbsent("order-1", "B", null), "B stores the key
first");
+ // the insert of A fails with a constraint violation
+ 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
putIfAbsentReturnsTheValueOfTheWinnerWhenTheDriverReportsOnlyTheSqlState()
throws Exception {
+ // as the PostgreSQL and SQL Server drivers do: a plain SQLException
with the SQL state of a unique violation
+ pausingA.reportConstraintViolationsBySqlState();
+ Gate beforeInsert = pausingA.pause(2);
+
+ Future<Object> a = executor.submit(() -> nodeA.putIfAbsent("order-2",
"A", null));
+ beforeInsert.awaitReached();
+ assertNull(nodeB.putIfAbsent("order-2", "B", null), "B stores the key
first");
+ beforeInsert.release();
+
+ assertEquals("B", a.get(10, TimeUnit.SECONDS), "A lost the race and
must get the value of B");
+ }
+
+ @Test
+ void putIfAbsentInsertsAgainWhenTheOtherEntryIsDeletedMeanwhile() throws
Exception {
+ Gate beforeInsert = pausingA.pause(2);
+ Gate beforeSecondRead = pausingA.pause(4);
+
+ 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 constraint violation, 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 putIfAbsentOnAnExpiredEntryHasOneWinner() throws Exception {
+ storeExpiredEntry("job-7");
+ Gate afterRead = pausingA.pause(2);
+
+ Future<Object> a = executor.submit(() -> nodeA.putIfAbsent("job-7",
"A", null));
+ // A has read the entry as expired
+ afterRead.awaitReached();
+ assertNull(nodeB.putIfAbsent("job-7", "B", null), "the expired entry
is absent, so B gets the key");
+ afterRead.release();
+
+ assertEquals("B", a.get(10, TimeUnit.SECONDS), "A lost the race and
must get the value of B");
+ assertEquals("B", nodeB.get("job-7"));
+ }
+
+ @Test
+ void readingAnExpiredEntryDoesNotDeleteTheEntryStoredMeanwhile() throws
Exception {
+ storeExpiredEntry("job-42");
+ Gate afterRead = pausingA.pause(2);
+
+ Future<Object> a = executor.submit(() -> nodeA.get("job-42"));
+ // A has read the entry as expired and is about to delete it
+ afterRead.awaitReached();
+ assertNull(nodeB.putIfAbsent("job-42", "B", null), "the expired entry
is absent, so B gets the key");
+ afterRead.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 putIfAbsentGivesUpAfterThreeConstraintViolations() {
+ assertNull(nodeB.putIfAbsent("job-1", "B", null));
+ // node A does not find the stored entry, so each of its inserts fails
with a constraint violation
+ pausingA.hideEntries();
+
+ assertThrows(PersistenceException.class, () ->
nodeA.putIfAbsent("job-1", "A", null));
+ assertEquals(3, pausingA.operations("persist"));
+ assertEquals("B", nodeB.get("job-1"));
+ }
+
+ private void storeExpiredEntry(String key) {
+ EntityManager em = entityManagerFactory.createEntityManager();
+ try {
+ em.getTransaction().begin();
+ // an entry that expired long ago
+ em.persist(new KeyValueEntry(key,
KeyValueRepositoryHelper.serialize("old"), 1L));
+ em.getTransaction().commit();
+ } finally {
+ em.close();
+ }
+ }
+
+ /**
+ * Pauses before the n-th entity manager operation of a {@link
PausingEntityManagerFactory}, until released.
+ */
+ private static final class Gate {
+ private final int occurrence;
+ private final CountDownLatch reached = new CountDownLatch(1);
+ private final CountDownLatch released = new CountDownLatch(1);
+
+ private Gate(int occurrence) {
+ this.occurrence = occurrence;
+ }
+
+ private void beforeOperation(int number) {
+ if (number == 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);
+ }
+ }
+ }
+
+ /**
+ * Wraps the entity managers of a factory, counts their operations that
read or write the database (query
+ * executions, persist, merge, remove, flush) and pauses them at the given
gates.
+ */
+ private static final class PausingEntityManagerFactory {
+ private final EntityManagerFactory delegate;
+ private final List<Gate> gates = new CopyOnWriteArrayList<>();
+ private final List<String> executed = new CopyOnWriteArrayList<>();
+ private final AtomicInteger count = new AtomicInteger();
+ private volatile boolean hideEntries;
+ private volatile boolean bySqlState;
+
+ private PausingEntityManagerFactory(EntityManagerFactory delegate) {
+ this.delegate = delegate;
+ }
+
+ Gate pause(int occurrence) {
+ Gate gate = new Gate(occurrence);
+ gates.add(gate);
+ return gate;
+ }
+
+ void hideEntries() {
+ hideEntries = true;
+ }
+
+ void reportConstraintViolationsBySqlState() {
+ bySqlState = true;
+ }
+
+ long operations(String name) {
+ return executed.stream().filter(name::equals).count();
+ }
+
+ void releaseAll() {
+ gates.forEach(Gate::release);
+ }
+
+ private void beforeOperation(String name) {
+ executed.add(name);
+ int number = count.incrementAndGet();
+ gates.forEach(gate -> gate.beforeOperation(number));
+ }
+
+ EntityManagerFactory proxy() {
+ return proxy(EntityManagerFactory.class, delegate, (method, args)
-> {
+ Object result = invoke(delegate, method, args);
+ if (result instanceof EntityManager em) {
+ return proxy(EntityManager.class, em, (m, a) ->
onEntityManager(em, m, a));
+ }
+ return result;
+ });
+ }
+
+ private Object onEntityManager(EntityManager em, Method method,
Object[] args) throws Throwable {
+ String name = method.getName();
+ if ("persist".equals(name) || "merge".equals(name) ||
"remove".equals(name) || "flush".equals(name)) {
+ beforeOperation(name);
+ }
+ Object result;
+ try {
+ result = invoke(em, method, args);
+ } catch (PersistenceException e) {
+ if (bySqlState && "flush".equals(name)) {
+ throw new PersistenceException(
+ "Simulated driver error",
+ new SQLException("duplicate key value violates
unique constraint", "23505"));
+ }
+ throw e;
+ }
+ if (result instanceof Query query && "createQuery".equals(name) &&
args[0] instanceof String jpql) {
+ return proxy(Query.class, query, (m, a) -> onQuery(query,
jpql, m, a));
+ }
+ return result;
+ }
+
+ private Object onQuery(Query query, String jpql, Method method,
Object[] args) throws Throwable {
+ String name = method.getName();
+ if ("getResultList".equals(name) || "getSingleResult".equals(name)
|| "executeUpdate".equals(name)) {
+ beforeOperation(jpql.substring(0, jpql.indexOf(' ')));
+ if (hideEntries && "getResultList".equals(name) &&
jpql.startsWith("select e from")) {
+ return Collections.emptyList();
+ }
+ }
+ return invoke(query, method, args);
+ }
+
+ private static Object invoke(Object target, Method method, Object[]
args) throws Throwable {
+ try {
+ return method.invoke(target, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ }
+
+ @FunctionalInterface
+ private interface Handler {
+ Object handle(Method method, Object[] args) throws Throwable;
+ }
+
+ private static <T> T proxy(Class<T> type, Object target, Handler
handler) {
+ InvocationHandler invocationHandler = (p, method, args) -> {
+ if (method.getDeclaringClass() == Object.class) {
+ // equals, hashCode and toString of the proxy are those of
the target
+ if ("equals".equals(method.getName())) {
+ return p == args[0];
+ }
+ return invoke(target, method, args);
+ }
+ return handler.handle(method, args);
+ };
+ return type.cast(Proxy.newProxyInstance(type.getClassLoader(), new
Class<?>[] { type }, invocationHandler));
+ }
+ }
+}
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 750a45c396ba..b7e26662cd75 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
@@ -80,7 +80,7 @@ Choosing the right backend is typically a matter of:
| `JpaKeyValueRepository`
| `camel-jpa`
| JPA entity-backed; works with any JPA 2 provider (Hibernate, EclipseLink…)
-| ✗ (non-atomic)
+| ✓ (primary key)
| ✗ (client-side)
| `CassandraKeyValueRepository`
@@ -675,8 +675,11 @@ YAML DSL::
----
====
-NOTE: The JPA backend does not provide atomic `putIfAbsent`. Idempotency under
concurrent
-load relies on unique-constraint violations being caught at the database level.
+NOTE: `JpaKeyValueRepository.putIfAbsent` relies on the primary key of the
table: when concurrent
+inserts of the same key fail with a constraint violation, the transaction is
rolled back and the key
+is read again in a new transaction, which returns the value stored by the
other caller. After three
+failed inserts the `PersistenceException` is thrown. When `putIfAbsent` runs
in a transaction of the
+caller, the failed insert marks that transaction for rollback, so the caller's
transaction fails.
=== 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 14c0c98d828d..7d1a8214d90f 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
@@ -7434,6 +7434,17 @@ The return type of the protected method
`JpaEndpoint.createTransactionStrategy()
of `DefaultTransactionStrategy`. A subclass that overrides it must be
recompiled: compiled against the old return type,
its method is no longer called.
+=== camel-jpa - JpaKeyValueRepository: putIfAbsent after a constraint violation
+
+When the insert of `JpaKeyValueRepository.putIfAbsent` fails with a constraint
violation because another caller stored
+the 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 the constraint violation had
marked for rollback, so `putIfAbsent`
+failed when that transaction was committed (with a
`DataIntegrityViolationException` when the transaction strategy is
+Spring based). After a constraint violation, `putIfAbsent` now throws a
`PersistenceException` only when the insert has
+failed three times. A constraint violation is also recognised by its SQL state
(class `23`), as the PostgreSQL and SQL
+Server drivers report it. When `putIfAbsent` runs in a transaction of the
caller, the failed insert marks that
+transaction for rollback, so the caller's transaction fails, as before.
+
=== camel-jbang - the export to Quarkus keeps camel-quarkus-microprofile-health
`camel export --runtime=quarkus`, and `camel run --runtime=quarkus`, which
exports the same way, now write
