This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24048 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 85d73dee7c3659073790cad5f6bee70aa30f00c2 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 11:07:32 2026 +0200 CAMEL-24048: camel-sql - Fix stale remove, idempotent race, and lexer error handling - JdbcAggregationRepository/ClusteredJdbcAggregationRepository remove() now checks the delete count and throws OptimisticLockingException on stale version, symmetric with updateHelper() - AbstractJdbcMessageIdRepository add() catches DuplicateKeyException from a concurrent insert race and returns false instead of propagating - TemplateParser catches TokenMgrError from the JavaCC lexer and wraps it in ParseRuntimeException instead of letting a raw Error escape Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../sql/stored/template/TemplateParser.java | 3 + .../jdbc/ClusteredJdbcAggregationRepository.java | 8 +- .../aggregate/jdbc/JdbcAggregationRepository.java | 8 +- .../jdbc/AbstractJdbcMessageIdRepository.java | 12 ++- .../sql/stored/TemplateParserLexicalErrorTest.java | 51 ++++++++++ .../JdbcAggregationRepositoryStaleRemoveTest.java | 52 +++++++++++ ...MessageIdRepositoryDuplicateInsertRaceTest.java | 104 +++++++++++++++++++++ 7 files changed, 233 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 800ab575fba2..01dc2e9d9071 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 @@ -437,12 +437,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..4e057c9f5c3a 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,14 @@ 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) { + // a concurrent insert won the race — the key already exists + 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; + } + } +}
