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 95074793d248 CAMEL-24048: camel-sql - Fix stale remove, idempotent
race, and lexer error handling
95074793d248 is described below
commit 95074793d2483aeb25b74b4299c016bd5290f17c
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 15:15:53 2026 +0200
CAMEL-24048: camel-sql - Fix stale remove, idempotent race, and lexer error
handling
Fix three correctness defects in camel-sql:
- JdbcAggregationRepository remove() now checks delete count and throws
OptimisticLockingException on stale version, preventing duplicate delivery
- AbstractJdbcMessageIdRepository.add() catches DuplicateKeyException from
concurrent insert race, returning false instead of propagating the error
- TemplateParser catches TokenMgrError from JavaCC lexer and wraps it in
ParseRuntimeException, consistent with ParseException handling
Closes #24710
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../sql/stored/template/TemplateParser.java | 3 +
.../jdbc/ClusteredJdbcAggregationRepository.java | 8 +-
.../aggregate/jdbc/JdbcAggregationRepository.java | 8 +-
.../jdbc/AbstractJdbcMessageIdRepository.java | 13 ++-
.../sql/stored/TemplateParserLexicalErrorTest.java | 51 ++++++++++
.../JdbcAggregationRepositoryStaleRemoveTest.java | 52 +++++++++++
...MessageIdRepositoryDuplicateInsertRaceTest.java | 104 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 12 +++
8 files changed, 246 insertions(+), 5 deletions(-)
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/component/sql/stored/template/TemplateParser.java
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/stored/template/TemplateParser.java
index 3a857653ddbd..2d82711f220b 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/component/sql/stored/template/TemplateParser.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/component/sql/stored/template/TemplateParser.java
@@ -22,6 +22,7 @@ import
org.apache.camel.component.sql.stored.template.ast.ParseRuntimeException;
import org.apache.camel.component.sql.stored.template.ast.Template;
import org.apache.camel.component.sql.stored.template.generated.ParseException;
import org.apache.camel.component.sql.stored.template.generated.SSPTParser;
+import org.apache.camel.component.sql.stored.template.generated.TokenMgrError;
import org.apache.camel.spi.ClassResolver;
import org.apache.camel.util.ObjectHelper;
@@ -41,6 +42,8 @@ public class TemplateParser {
} catch (ParseException parseException) {
throw new ParseRuntimeException(parseException);
+ } catch (TokenMgrError tokenMgrError) {
+ throw new ParseRuntimeException(tokenMgrError);
}
}
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredJdbcAggregationRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredJdbcAggregationRepository.java
index 28af65e419e8..135d9efedec9 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredJdbcAggregationRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredJdbcAggregationRepository.java
@@ -77,11 +77,17 @@ public class ClusteredJdbcAggregationRepository extends
JdbcAggregationRepositor
LOG.debug("Removing key {}", correlationId);
String table = getRepositoryName();
verifyTableName(table);
- jdbcTemplate.update("DELETE FROM " + table + " WHERE " +
ID + " = ? AND " + VERSION + " = ?", // NOSONAR
+ int deleteCount = jdbcTemplate.update(
+ "DELETE FROM " + table + " WHERE " + ID + " = ?
AND " + VERSION + " = ?", // NOSONAR
correlationId, version);
+ if (deleteCount != 1) {
+ throw new OptimisticLockingException();
+ }
insert(camelContext, confirmKey, exchange,
getRepositoryNameCompleted(), version, true);
+ } catch (OptimisticLockingException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException(
"Error removing key " + correlationId + " from
repository " + getRepositoryName(), e);
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepository.java
index 0593b7f71227..7d401ad90490 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepository.java
@@ -438,12 +438,18 @@ public class JdbcAggregationRepository extends
ServiceSupport
LOG.debug("Removing key {}", correlationId);
String table = getRepositoryName();
verifyTableName(table);
- jdbcTemplate.update("DELETE FROM " + table + " WHERE " +
ID + " = ? AND " + VERSION + " = ?", // NOSONAR
+ int deleteCount = jdbcTemplate.update(
+ "DELETE FROM " + table + " WHERE " + ID + " = ?
AND " + VERSION + " = ?", // NOSONAR
correlationId, version);
+ if (deleteCount != 1) {
+ throw new OptimisticLockingException();
+ }
insert(camelContext, confirmKey, exchange,
getRepositoryNameCompleted(), version);
LOG.debug("Removed key {}", correlationId);
+ } catch (OptimisticLockingException e) {
+ throw e;
} catch (Exception e) {
throw new RuntimeException("Error removing key " +
correlationId + " from repository " + repositoryName, e);
}
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/AbstractJdbcMessageIdRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/AbstractJdbcMessageIdRepository.java
index 63f5353074da..dc5360ed29c1 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/AbstractJdbcMessageIdRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/AbstractJdbcMessageIdRepository.java
@@ -25,6 +25,7 @@ import org.apache.camel.spi.Metadata;
import org.apache.camel.support.service.ServiceSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.dao.DuplicateKeyException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;
@@ -140,9 +141,15 @@ public abstract class AbstractJdbcMessageIdRepository
extends ServiceSupport imp
public Boolean doInTransaction(TransactionStatus status) {
int count = queryForInt(key);
if (count == 0) {
- int insertedCount = insert(key);
- if (insertedCount != 0) {
- return Boolean.TRUE;
+ try {
+ int insertedCount = insert(key);
+ if (insertedCount != 0) {
+ return Boolean.TRUE;
+ }
+ } catch (DuplicateKeyException e) {
+ log.debug("Concurrent insert race for key '{}' —
another node/thread won, treating as duplicate", key);
+ status.setRollbackOnly();
+ return Boolean.FALSE;
}
}
return Boolean.FALSE;
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/component/sql/stored/TemplateParserLexicalErrorTest.java
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/stored/TemplateParserLexicalErrorTest.java
new file mode 100644
index 000000000000..ee0ddaf4a55f
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/stored/TemplateParserLexicalErrorTest.java
@@ -0,0 +1,51 @@
+/*
+ * 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.component.sql.stored;
+
+import org.apache.camel.component.sql.stored.template.TemplateParser;
+import
org.apache.camel.component.sql.stored.template.ast.ParseRuntimeException;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Verifies that a lexical error (character outside the token alphabet) in a
stored procedure template is wrapped in
+ * {@link ParseRuntimeException} rather than escaping as a raw {@link Error}.
+ */
+public class TemplateParserLexicalErrorTest extends CamelTestSupport {
+
+ TemplateParser parser;
+
+ @BeforeEach
+ void setupTest() {
+ parser = new TemplateParser(context.getClassResolver());
+ }
+
+ @Test
+ void testSemicolonThrowsParseRuntimeException() {
+ assertThrows(ParseRuntimeException.class,
+ () -> parser.parseTemplate("MYFUNC(INTEGER ${header.foo});"));
+ }
+
+ @Test
+ void testBacktickThrowsParseRuntimeException() {
+ assertThrows(ParseRuntimeException.class,
+ () -> parser.parseTemplate("MYFUNC`(INTEGER ${header.foo})"));
+ }
+}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepositoryStaleRemoveTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepositoryStaleRemoveTest.java
new file mode 100644
index 000000000000..6058b020d59a
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepositoryStaleRemoveTest.java
@@ -0,0 +1,52 @@
+/*
+ * 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.aggregate.jdbc;
+
+import org.apache.camel.Exchange;
+import
org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class JdbcAggregationRepositoryStaleRemoveTest extends
AbstractJdbcAggregationTestSupport {
+
+ @Override
+ void configureJdbcAggregationRepository() {
+ super.configureJdbcAggregationRepository();
+ repo.setReturnOldExchange(true);
+ }
+
+ @Test
+ public void testStaleRemoveThrowsOptimisticLockingException() {
+ Exchange exchange1 = new DefaultExchange(context);
+ exchange1.getIn().setBody("body1");
+ repo.add(context, "foo", exchange1);
+
+ // get exchange with version 1
+ Exchange staleExchange = repo.get(context, "foo");
+
+ // add again to bump the version in the database
+ Exchange exchange2 = repo.get(context, "foo");
+ exchange2.getIn().setBody("body2");
+ repo.add(context, "foo", exchange2);
+
+ // staleExchange still carries the old version — remove must detect
the mismatch
+ assertThrows(OptimisticLockingException.class,
+ () -> repo.remove(context, "foo", staleExchange));
+ }
+}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcMessageIdRepositoryDuplicateInsertRaceTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcMessageIdRepositoryDuplicateInsertRaceTest.java
new file mode 100644
index 000000000000..cd13a93f4635
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcMessageIdRepositoryDuplicateInsertRaceTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.idempotent.jdbc;
+
+import javax.sql.DataSource;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies that a concurrent duplicate insert in add() is treated as "already
exists" (returns false) instead of
+ * propagating a DuplicateKeyException.
+ * <p>
+ * The race is simulated by overriding queryForInt() to return 0 even when the
key is already present, forcing the
+ * INSERT to hit the primary-key constraint.
+ */
+public class JdbcMessageIdRepositoryDuplicateInsertRaceTest {
+
+ private static final String PROCESSOR_NAME = "testProcessor";
+
+ private EmbeddedDatabase dataSource;
+ private JdbcMessageIdRepository repo;
+
+ @BeforeEach
+ void setUp() {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .setName("idempotent-race-" + System.identityHashCode(this))
+ .build();
+
+ repo = new RaceSimulatingRepository(dataSource, PROCESSOR_NAME);
+ repo.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ repo.stop();
+ dataSource.shutdown();
+ }
+
+ @Test
+ void testDuplicateInsertReturnsFalseInsteadOfThrowing() {
+ // first add succeeds normally
+ assertTrue(repo.add("key1"));
+
+ // second add hits the PK constraint because queryForInt() is
overridden to return 0;
+ // before the fix this throws DuplicateKeyException; after the fix it
returns false
+ assertDoesNotThrow(() -> {
+ boolean result = repo.add("key1");
+ assertFalse(result, "add() should return false when a concurrent
insert already inserted the key");
+ });
+ }
+
+ /**
+ * Subclass that overrides queryForInt() to always return 0 after the
first successful add, simulating a concurrent
+ * node that inserted the same key between the SELECT COUNT(*) and INSERT
statements.
+ */
+ static class RaceSimulatingRepository extends JdbcMessageIdRepository {
+
+ private volatile boolean simulateRace;
+
+ RaceSimulatingRepository(DataSource dataSource, String processorName) {
+ super(dataSource, processorName);
+ }
+
+ @Override
+ protected int queryForInt(String key) {
+ if (simulateRace) {
+ return 0;
+ }
+ return super.queryForInt(key);
+ }
+
+ @Override
+ protected int insert(String key) {
+ // after the first successful insert, enable race simulation
+ int result = super.insert(key);
+ simulateRace = true;
+ return result;
+ }
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 61e5b7cd109c..dfdcc67db020 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -455,3 +455,15 @@ The table-name validation in `JdbcAggregationRepository`
now accepts schema-qual
such as `myschema.aggregation`. Previously the validation regex only allowed
simple identifiers
(`[a-zA-Z_][a-zA-Z0-9_]*`), rejecting any name containing a dot. Names
starting with a digit,
containing spaces, or with multiple dots (e.g. `catalog.schema.table`) are
still rejected.
+
+=== camel-sql - New exceptions from aggregation and template parsing
+
+The `remove()` method in `JdbcAggregationRepository` and
`ClusteredJdbcAggregationRepository`
+now throws `OptimisticLockingException` when it detects a stale version during
the delete.
+Previously a stale remove was silently treated as successful. If your error
handling or
+aggregation strategy catches specific exception types around `remove()`, you
may need to
+account for `OptimisticLockingException`.
+
+The `TemplateParser` now catches `TokenMgrError` (a JavaCC lexer error) and
wraps it in
+`ParseRuntimeException`. Previously a malformed stored-procedure template with
characters
+outside the token alphabet would propagate as a raw `java.lang.Error`.
