This is an automated email from the ASF dual-hosted git repository.
Croway 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 6449c0ee3de2 CAMEL-24046: camel-sql - JdbcCachedMessageIdRepository
must not cache a key whose insert failed, make cache thread-safe
6449c0ee3de2 is described below
commit 6449c0ee3de2ccd38f8f583191fef59ab79eef73
Author: croway <[email protected]>
AuthorDate: Mon Jul 13 11:42:33 2026 +0200
CAMEL-24046: camel-sql - JdbcCachedMessageIdRepository must not cache a key
whose insert failed, make cache thread-safe
---
.../jdbc/JdbcCachedMessageIdRepository.java | 35 +++++----
...cCachedMessageIdRepositoryFailedInsertTest.java | 85 ++++++++++++++++++++++
2 files changed, 105 insertions(+), 15 deletions(-)
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
index eac9de9c1b16..dbb4f010adbd 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepository.java
@@ -16,8 +16,9 @@
*/
package org.apache.camel.processor.idempotent.jdbc;
-import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import javax.sql.DataSource;
@@ -29,9 +30,9 @@ import
org.springframework.transaction.support.TransactionTemplate;
* Caching version of {@link JdbcMessageIdRepository}
*/
public class JdbcCachedMessageIdRepository extends JdbcMessageIdRepository {
- private Map<String, Integer> cache = new HashMap<>();
- private int hitCount;
- private int missCount;
+ private volatile Map<String, Integer> cache = new ConcurrentHashMap<>();
+ private final AtomicInteger hitCount = new AtomicInteger();
+ private final AtomicInteger missCount = new AtomicInteger();
private String queryAllString
= "SELECT messageId, COUNT(*) FROM CAMEL_MESSAGEPROCESSED WHERE
processorName = ? GROUP BY messageId";
@@ -67,23 +68,27 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
@Override
public boolean add(final String key) {
Integer previousValue = cache.getOrDefault(key, 0);
- cache.put(key, previousValue + 1);
if (previousValue != 0) {
- hitCount++;
+ cache.merge(key, 1, Integer::sum);
+ hitCount.incrementAndGet();
return false;
}
- missCount++;
- return super.add(key);
+ missCount.incrementAndGet();
+ boolean added = super.add(key);
+ // only remember the key after the database insert succeeded,
otherwise a failed insert
+ // would leave the key cached and every redelivery would be rejected
as a duplicate
+ cache.merge(key, 1, Integer::sum);
+ return added;
}
@Override
public boolean contains(final String key) {
Integer previousValue = cache.getOrDefault(key, 0);
if (previousValue != 0) {
- hitCount++;
+ hitCount.incrementAndGet();
return true;
}
- missCount++;
+ missCount.incrementAndGet();
return super.contains(key);
}
@@ -96,8 +101,8 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
@Override
public void clear() {
cache.clear();
- hitCount = 0;
- missCount = 0;
+ hitCount.set(0);
+ missCount.set(0);
super.clear();
}
@@ -110,18 +115,18 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
}
public int getHitCount() {
- return hitCount;
+ return hitCount.get();
}
public int getMissCount() {
- return missCount;
+ return missCount.get();
}
public void reload() {
transactionTemplate.execute(status -> {
try {
cache = jdbcTemplate.query(getQueryAllString(), resultSet -> {
- Map<String, Integer> messageIdCount = new HashMap<>();
+ Map<String, Integer> messageIdCount = new
ConcurrentHashMap<>();
while (resultSet.next()) {
messageIdCount.put(resultSet.getString(1),
resultSet.getInt(2));
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryFailedInsertTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryFailedInsertTest.java
new file mode 100644
index 000000000000..2e3c63e4d087
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryFailedInsertTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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 org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * When the database insert fails, {@link
JdbcCachedMessageIdRepository#add(String)} propagates the exception and the
+ * idempotent consumer EIP never processes the message. The failed key must
therefore not be remembered as processed: a
+ * redelivery of the same message must be processable once the database is
available again.
+ */
+public class JdbcCachedMessageIdRepositoryFailedInsertTest {
+
+ private static final String PROCESSOR_NAME = "myProcessorName";
+
+ private EmbeddedDatabase dataSource;
+ private JdbcTemplate jdbcTemplate;
+ private JdbcCachedMessageIdRepository repository;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcTemplate = new JdbcTemplate(dataSource);
+ repository = new JdbcCachedMessageIdRepository(dataSource,
PROCESSOR_NAME);
+ repository.start();
+ }
+
+ @AfterEach
+ public void tearDown() throws Exception {
+ if (repository != null) {
+ repository.stop();
+ }
+ if (dataSource != null) {
+ dataSource.shutdown();
+ }
+ }
+
+ @Test
+ public void failedInsertMustNotMarkMessageAsProcessed() {
+ // sanity: normal add works and duplicates are detected
+ assertTrue(repository.add("first"));
+ assertFalse(repository.add("first"));
+
+ // simulate a database outage while adding a new key
+ jdbcTemplate.execute("DROP TABLE CAMEL_MESSAGEPROCESSED");
+ assertThrows(Exception.class, () -> repository.add("second"));
+
+ // database is back
+ jdbcTemplate.execute(repository.getCreateString());
+
+ // the failed add must not have marked "second" as processed: the
message was
+ // never processed and nothing was stored in the database
+ assertFalse(repository.contains("second"),
+ "a key whose insert failed must not be reported as already
processed");
+ assertTrue(repository.add("second"),
+ "redelivery of a message whose insert failed must be
processable");
+ }
+}
