This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24132 in repository https://gitbox.apache.org/repos/asf/camel.git
commit b713b678280d70b8417ee18111cc5bf794e3c748 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Jul 16 14:22:52 2026 +0200 CAMEL-24132: camel-jpa - Fix medium-severity findings from component review - 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 Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/jpa/JpaPollingConsumer.java | 23 ++++++++++++------- .../apache/camel/component/jpa/JpaProducer.java | 19 ++++++++-------- .../idempotent/jpa/JpaMessageIdRepository.java | 26 +++++++++++++++++++--- 3 files changed, 47 insertions(+), 21 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 0564411a47c1..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,6 +124,16 @@ public class JpaPollingConsumer extends PollingConsumerSupport { @Override public Exchange receive() { + 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(null, entityManagerFactory, getEndpoint().isUsePassedInEntityManager(), getEndpoint().isSharedEntityManager(), true); @@ -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 0194d3218c12..dd0b307727d9 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()) { @@ -301,6 +307,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
