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 <[email protected]>
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 <[email protected]>
- 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 <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>
---
.../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
