Skip to content

Commit 9507479

Browse files
davsclausclaude
davsclaus
and
claude
authored
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 <noreply@anthropic.com>
1 parent 7eb935d commit 9507479

8 files changed

Lines changed: 246 additions & 5 deletions

File tree

‎components/camel-sql/src/main/java/org/apache/camel/component/sql/stored/template/TemplateParser.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.camel.component.sql.stored.template.ast.Template;
2323
import org.apache.camel.component.sql.stored.template.generated.ParseException;
2424
import org.apache.camel.component.sql.stored.template.generated.SSPTParser;
25+
import org.apache.camel.component.sql.stored.template.generated.TokenMgrError;
2526
import org.apache.camel.spi.ClassResolver;
2627
import org.apache.camel.util.ObjectHelper;
2728

@@ -41,6 +42,8 @@ public Template parseTemplate(String template) {
4142

4243
} catch (ParseException parseException) {
4344
throw new ParseRuntimeException(parseException);
45+
} catch (TokenMgrError tokenMgrError) {
46+
throw new ParseRuntimeException(tokenMgrError);
4447
}
4548
}
4649

‎components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredJdbcAggregationRepository.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,11 +77,17 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
7777
LOG.debug("Removing key {}", correlationId);
7878
String table = getRepositoryName();
7979
verifyTableName(table);
80-
jdbcTemplate.update("DELETE FROM " + table + " WHERE " + ID + " = ? AND " + VERSION + " = ?", // NOSONAR
80+
int deleteCount = jdbcTemplate.update(
81+
"DELETE FROM " + table + " WHERE " + ID + " = ? AND " + VERSION + " = ?", // NOSONAR
8182
correlationId, version);
83+
if (deleteCount != 1) {
84+
throw new OptimisticLockingException();
85+
}
8286

8387
insert(camelContext, confirmKey, exchange, getRepositoryNameCompleted(), version, true);
8488

89+
} catch (OptimisticLockingException e) {
90+
throw e;
8591
} catch (Exception e) {
8692
throw new RuntimeException(
8793
"Error removing key " + correlationId + " from repository " + getRepositoryName(), e);

‎components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepository.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -438,12 +438,18 @@ protected void doInTransactionWithoutResult(TransactionStatus status) {
438438
LOG.debug("Removing key {}", correlationId);
439439
String table = getRepositoryName();
440440
verifyTableName(table);
441-
jdbcTemplate.update("DELETE FROM " + table + " WHERE " + ID + " = ? AND " + VERSION + " = ?", // NOSONAR
441+
int deleteCount = jdbcTemplate.update(
442+
"DELETE FROM " + table + " WHERE " + ID + " = ? AND " + VERSION + " = ?", // NOSONAR
442443
correlationId, version);
444+
if (deleteCount != 1) {
445+
throw new OptimisticLockingException();
446+
}
443447

444448
insert(camelContext, confirmKey, exchange, getRepositoryNameCompleted(), version);
445449
LOG.debug("Removed key {}", correlationId);
446450

451+
} catch (OptimisticLockingException e) {
452+
throw e;
447453
} catch (Exception e) {
448454
throw new RuntimeException("Error removing key " + correlationId + " from repository " + repositoryName, e);
449455
}

‎components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/AbstractJdbcMessageIdRepository.java‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.camel.support.service.ServiceSupport;
2626
import org.slf4j.Logger;
2727
import org.slf4j.LoggerFactory;
28+
import org.springframework.dao.DuplicateKeyException;
2829
import org.springframework.jdbc.core.JdbcTemplate;
2930
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
3031
import org.springframework.transaction.TransactionDefinition;
@@ -140,9 +141,15 @@ public boolean add(final String key) {
140141
public Boolean doInTransaction(TransactionStatus status) {
141142
int count = queryForInt(key);
142143
if (count == 0) {
143-
int insertedCount = insert(key);
144-
if (insertedCount != 0) {
145-
return Boolean.TRUE;
144+
try {
145+
int insertedCount = insert(key);
146+
if (insertedCount != 0) {
147+
return Boolean.TRUE;
148+
}
149+
} catch (DuplicateKeyException e) {
150+
log.debug("Concurrent insert race for key '{}' — another node/thread won, treating as duplicate", key);
151+
status.setRollbackOnly();
152+
return Boolean.FALSE;
146153
}
147154
}
148155
return Boolean.FALSE;

‎components/camel-sql/src/test/java/org/apache/camel/component/sql/stored/TemplateParserLexicalErrorTest.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.camel.component.sql.stored;
18+
19+
import org.apache.camel.component.sql.stored.template.TemplateParser;
20+
import org.apache.camel.component.sql.stored.template.ast.ParseRuntimeException;
21+
import org.apache.camel.test.junit6.CamelTestSupport;
22+
import org.junit.jupiter.api.BeforeEach;
23+
import org.junit.jupiter.api.Test;
24+
25+
import static org.junit.jupiter.api.Assertions.assertThrows;
26+
27+
/**
28+
* Verifies that a lexical error (character outside the token alphabet) in a stored procedure template is wrapped in
29+
* {@link ParseRuntimeException} rather than escaping as a raw {@link Error}.
30+
*/
31+
public class TemplateParserLexicalErrorTest extends CamelTestSupport {
32+
33+
TemplateParser parser;
34+
35+
@BeforeEach
36+
void setupTest() {
37+
parser = new TemplateParser(context.getClassResolver());
38+
}
39+
40+
@Test
41+
void testSemicolonThrowsParseRuntimeException() {
42+
assertThrows(ParseRuntimeException.class,
43+
() -> parser.parseTemplate("MYFUNC(INTEGER ${header.foo});"));
44+
}
45+
46+
@Test
47+
void testBacktickThrowsParseRuntimeException() {
48+
assertThrows(ParseRuntimeException.class,
49+
() -> parser.parseTemplate("MYFUNC`(INTEGER ${header.foo})"));
50+
}
51+
}

‎components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/JdbcAggregationRepositoryStaleRemoveTest.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.camel.processor.aggregate.jdbc;
18+
19+
import org.apache.camel.Exchange;
20+
import org.apache.camel.spi.OptimisticLockingAggregationRepository.OptimisticLockingException;
21+
import org.apache.camel.support.DefaultExchange;
22+
import org.junit.jupiter.api.Test;
23+
24+
import static org.junit.jupiter.api.Assertions.assertThrows;
25+
26+
public class JdbcAggregationRepositoryStaleRemoveTest extends AbstractJdbcAggregationTestSupport {
27+
28+
@Override
29+
void configureJdbcAggregationRepository() {
30+
super.configureJdbcAggregationRepository();
31+
repo.setReturnOldExchange(true);
32+
}
33+
34+
@Test
35+
public void testStaleRemoveThrowsOptimisticLockingException() {
36+
Exchange exchange1 = new DefaultExchange(context);
37+
exchange1.getIn().setBody("body1");
38+
repo.add(context, "foo", exchange1);
39+
40+
// get exchange with version 1
41+
Exchange staleExchange = repo.get(context, "foo");
42+
43+
// add again to bump the version in the database
44+
Exchange exchange2 = repo.get(context, "foo");
45+
exchange2.getIn().setBody("body2");
46+
repo.add(context, "foo", exchange2);
47+
48+
// staleExchange still carries the old version — remove must detect the mismatch
49+
assertThrows(OptimisticLockingException.class,
50+
() -> repo.remove(context, "foo", staleExchange));
51+
}
52+
}

‎components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcMessageIdRepositoryDuplicateInsertRaceTest.java‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package org.apache.camel.processor.idempotent.jdbc;
18+
19+
import javax.sql.DataSource;
20+
21+
import org.junit.jupiter.api.AfterEach;
22+
import org.junit.jupiter.api.BeforeEach;
23+
import org.junit.jupiter.api.Test;
24+
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
25+
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
26+
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
27+
28+
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
29+
import static org.junit.jupiter.api.Assertions.assertFalse;
30+
import static org.junit.jupiter.api.Assertions.assertTrue;
31+
32+
/**
33+
* Verifies that a concurrent duplicate insert in add() is treated as "already exists" (returns false) instead of
34+
* propagating a DuplicateKeyException.
35+
* <p>
36+
* The race is simulated by overriding queryForInt() to return 0 even when the key is already present, forcing the
37+
* INSERT to hit the primary-key constraint.
38+
*/
39+
public class JdbcMessageIdRepositoryDuplicateInsertRaceTest {
40+
41+
private static final String PROCESSOR_NAME = "testProcessor";
42+
43+
private EmbeddedDatabase dataSource;
44+
private JdbcMessageIdRepository repo;
45+
46+
@BeforeEach
47+
void setUp() {
48+
dataSource = new EmbeddedDatabaseBuilder()
49+
.setType(EmbeddedDatabaseType.H2)
50+
.setName("idempotent-race-" + System.identityHashCode(this))
51+
.build();
52+
53+
repo = new RaceSimulatingRepository(dataSource, PROCESSOR_NAME);
54+
repo.start();
55+
}
56+
57+
@AfterEach
58+
void tearDown() {
59+
repo.stop();
60+
dataSource.shutdown();
61+
}
62+
63+
@Test
64+
void testDuplicateInsertReturnsFalseInsteadOfThrowing() {
65+
// first add succeeds normally
66+
assertTrue(repo.add("key1"));
67+
68+
// second add hits the PK constraint because queryForInt() is overridden to return 0;
69+
// before the fix this throws DuplicateKeyException; after the fix it returns false
70+
assertDoesNotThrow(() -> {
71+
boolean result = repo.add("key1");
72+
assertFalse(result, "add() should return false when a concurrent insert already inserted the key");
73+
});
74+
}
75+
76+
/**
77+
* Subclass that overrides queryForInt() to always return 0 after the first successful add, simulating a concurrent
78+
* node that inserted the same key between the SELECT COUNT(*) and INSERT statements.
79+
*/
80+
static class RaceSimulatingRepository extends JdbcMessageIdRepository {
81+
82+
private volatile boolean simulateRace;
83+
84+
RaceSimulatingRepository(DataSource dataSource, String processorName) {
85+
super(dataSource, processorName);
86+
}
87+
88+
@Override
89+
protected int queryForInt(String key) {
90+
if (simulateRace) {
91+
return 0;
92+
}
93+
return super.queryForInt(key);
94+
}
95+
96+
@Override
97+
protected int insert(String key) {
98+
// after the first successful insert, enable race simulation
99+
int result = super.insert(key);
100+
simulateRace = true;
101+
return result;
102+
}
103+
}
104+
}

‎docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -455,3 +455,15 @@ The table-name validation in `JdbcAggregationRepository` now accepts schema-qual
455455
such as `myschema.aggregation`. Previously the validation regex only allowed simple identifiers
456456
(`[a-zA-Z_][a-zA-Z0-9_]*`), rejecting any name containing a dot. Names starting with a digit,
457457
containing spaces, or with multiple dots (e.g. `catalog.schema.table`) are still rejected.
458+
459+
=== camel-sql - New exceptions from aggregation and template parsing
460+
461+
The `remove()` method in `JdbcAggregationRepository` and `ClusteredJdbcAggregationRepository`
462+
now throws `OptimisticLockingException` when it detects a stale version during the delete.
463+
Previously a stale remove was silently treated as successful. If your error handling or
464+
aggregation strategy catches specific exception types around `remove()`, you may need to
465+
account for `OptimisticLockingException`.
466+
467+
The `TemplateParser` now catches `TokenMgrError` (a JavaCC lexer error) and wraps it in
468+
`ParseRuntimeException`. Previously a malformed stored-procedure template with characters
469+
outside the token alphabet would propagate as a raw `java.lang.Error`.

0 commit comments

Comments
 (0)