You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/16 15:07:24 UTC
(camel) branch main updated: CAMEL-24132: camel-jpa - Fix medium-severity findings from component review
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 9e2460306476 CAMEL-24132: camel-jpa - Fix medium-severity findings from component review
9e2460306476 is described below
commit 9e2460306476245cd61f7f1a3da64d455ef61972
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Jul 16 17:07:18 2026 +0200
CAMEL-24132: camel-jpa - Fix medium-severity findings from component review
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
- M1: receiveNoWait() no longer delegates to receive() with pessimistic locking
- M2: receive(timeout) cancels the background future on timeout
- M3: setLockMode() is skipped for native queries per JPA spec
- M4: Concurrent insert race in JpaMessageIdRepository maps constraint violation to duplicate
- M6: isUseExecuteUpdate() trims query and detects update keywords instead of select
Signed-off-by: Claus Ibsen <cl...@gmail.com>
Co-authored-by: Claude Opus 4.6 <no...@anthropic.com>
---
.../camel/component/jpa/JpaPollingConsumer.java | 27 ++++++++++++++--------
.../apache/camel/component/jpa/JpaProducer.java | 19 ++++++++-------
.../idempotent/jpa/JpaMessageIdRepository.java | 26 ++++++++++++++++++---
3 files changed, 49 insertions(+), 23 deletions(-)
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaPollingConsumer.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaPollingConsumer.java
index d0391956d87b..3e327fd384b0 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaPollingConsumer.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaPollingConsumer.java
@@ -124,11 +124,21 @@ public class JpaPollingConsumer extends PollingConsumerSupport {
@Override
public Exchange receive() {
- Exchange exchange = getEndpoint().createExchange();
+ return doReceive(true);
+ }
+
+ @Override
+ public Exchange receiveNoWait() {
+ // do not use locking so the call returns immediately without blocking on row locks
+ return doReceive(false);
+ }
+ private Exchange doReceive(boolean useLocking) {
// resolve the entity manager before evaluating the expression
- final EntityManager entityManager = getTargetEntityManager(exchange, entityManagerFactory,
+ final EntityManager entityManager = getTargetEntityManager(null, entityManagerFactory,
getEndpoint().isUsePassedInEntityManager(), getEndpoint().isSharedEntityManager(), true);
+
+ Exchange exchange = getEndpoint().createExchange();
exchange.getIn().setHeader(JpaConstants.ENTITY_MANAGER, entityManager);
transactionStrategy.executeInTransaction(new Runnable() {
@Override
@@ -140,7 +150,9 @@ public class JpaPollingConsumer extends PollingConsumerSupport {
Query innerQuery = getQueryFactory().createQuery(entityManager);
configureParameters(innerQuery);
- if (getEndpoint().isConsumeLockEntity()) {
+ // only apply lock mode when locking is enabled and not using native queries
+ // as JPA spec does not support setLockMode on native queries
+ if (useLocking && getEndpoint().isConsumeLockEntity() && nativeQuery == null) {
innerQuery.setLockMode(getLockModeType());
}
@@ -182,12 +194,6 @@ public class JpaPollingConsumer extends PollingConsumerSupport {
return exchange;
}
- @Override
- public Exchange receiveNoWait() {
- // call receive as-is
- return receive();
- }
-
@Override
public Exchange receive(long timeout) {
// need to use a thread pool to perform the task so we can support timeout
@@ -205,7 +211,8 @@ public class JpaPollingConsumer extends PollingConsumerSupport {
} catch (ExecutionException e) {
throw RuntimeCamelException.wrapRuntimeCamelException(e);
} catch (TimeoutException e) {
- // ignore as we hit timeout then return null
+ // cancel the task so it does not keep running in the background holding DB locks
+ future.cancel(true);
}
return null;
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
index ad83553046f0..b0b0ac831aa3 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaProducer.java
@@ -145,17 +145,9 @@ public class JpaProducer extends DefaultProducer {
public boolean isUseExecuteUpdate() {
if (useExecuteUpdate == null) {
if (query != null) {
- if (query.regionMatches(true, 0, "select", 0, 6)) {
- useExecuteUpdate = false;
- } else {
- useExecuteUpdate = true;
- }
+ useExecuteUpdate = isUpdateQuery(query);
} else if (nativeQuery != null) {
- if (nativeQuery.regionMatches(true, 0, "select", 0, 6)) {
- useExecuteUpdate = false;
- } else {
- useExecuteUpdate = true;
- }
+ useExecuteUpdate = isUpdateQuery(nativeQuery);
} else {
useExecuteUpdate = false;
}
@@ -163,6 +155,13 @@ public class JpaProducer extends DefaultProducer {
return useExecuteUpdate;
}
+ private static boolean isUpdateQuery(String queryText) {
+ String trimmed = queryText.stripLeading();
+ return trimmed.regionMatches(true, 0, "insert", 0, 6)
+ || trimmed.regionMatches(true, 0, "update", 0, 6)
+ || trimmed.regionMatches(true, 0, "delete", 0, 6);
+ }
+
@Override
public void process(final Exchange exchange) {
// resolve the entity manager before evaluating the expression
diff --git a/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java b/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
index 30f6d25a1d47..d9eb966abb62 100644
--- a/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
+++ b/components/camel-jpa/src/main/java/org/apache/camel/processor/idempotent/jpa/JpaMessageIdRepository.java
@@ -107,14 +107,20 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
processed.setCreatedAt(new Date());
entityManager.persist(processed);
entityManager.flush();
- entityManager.close();
rc[0] = Boolean.TRUE;
} else {
rc[0] = Boolean.FALSE;
}
} catch (Exception ex) {
- String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage());
- throw new PersistenceException(contextInfo, ex);
+ if (isConstraintViolation(ex)) {
+ // concurrent insert of the same messageId detected via unique constraint
+ // treat as duplicate rather than failing the exchange
+ LOG.debug("Duplicate message detected during concurrent insert: {}", messageId);
+ rc[0] = Boolean.FALSE;
+ } else {
+ String contextInfo = String.format(SOMETHING_WENT_WRONG, ex.getMessage());
+ throw new PersistenceException(contextInfo, ex);
+ }
} finally {
try {
if (entityManager.isOpen()) {
@@ -302,6 +308,20 @@ public class JpaMessageIdRepository extends ServiceSupport implements Idempotent
this.sharedEntityManager = sharedEntityManager;
}
+ private static boolean isConstraintViolation(Exception ex) {
+ Throwable cause = ex;
+ while (cause != null) {
+ if (cause instanceof java.sql.SQLIntegrityConstraintViolationException) {
+ return true;
+ }
+ if (cause instanceof jakarta.persistence.EntityExistsException) {
+ return true;
+ }
+ cause = cause.getCause();
+ }
+ return false;
+ }
+
@Override
protected void doStart() throws Exception {
// noop