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 6c100bb4d822 CAMEL-26705: camel-sql -
JdbcOrphanLockAwareIdempotentRepository and JdbcCachedMessageIdRepository are
not restart safe (#29402)
6c100bb4d822 is described below
commit 6c100bb4d82234b093c1d07eb7ec85861e64a5a6
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 13:32:32 2026 +0530
CAMEL-26705: camel-sql - JdbcOrphanLockAwareIdempotentRepository and
JdbcCachedMessageIdRepository are not restart safe (#29402)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../jdbc/JdbcCachedMessageIdRepository.java | 10 +-
.../JdbcOrphanLockAwareIdempotentRepository.java | 19 ++-
.../JdbcCachedMessageIdRepositoryRestartTest.java | 82 +++++++++++++
...anLockAwareIdempotentRepositoryRestartTest.java | 127 +++++++++++++++++++++
4 files changed, 231 insertions(+), 7 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 dbb4f010adbd..07c2bbb91f42 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
@@ -30,11 +30,14 @@ import
org.springframework.transaction.support.TransactionTemplate;
* Caching version of {@link JdbcMessageIdRepository}
*/
public class JdbcCachedMessageIdRepository extends JdbcMessageIdRepository {
+
+ protected static final String DEFAULT_QUERY_ALL_STRING
+ = "SELECT messageId, COUNT(*) FROM CAMEL_MESSAGEPROCESSED WHERE
processorName = ? GROUP BY messageId";
+
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";
+ private String queryAllString = DEFAULT_QUERY_ALL_STRING;
public JdbcCachedMessageIdRepository() {
}
@@ -55,7 +58,8 @@ public class JdbcCachedMessageIdRepository extends
JdbcMessageIdRepository {
protected void doInit() throws Exception {
super.doInit();
if (getTableName() != null) {
- queryAllString = queryAllString.replaceFirst(DEFAULT_TABLENAME,
getTableName());
+ // update query string from default table name to the new table
name
+ queryAllString =
DEFAULT_QUERY_ALL_STRING.replace(DEFAULT_TABLENAME, getTableName());
}
}
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
index 0cb53e1e4388..10bcaee14429 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepository.java
@@ -46,6 +46,9 @@ import
org.springframework.transaction.support.TransactionTemplate;
*/
public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdRepository {
+ protected static final String DEFAULT_UPDATE_TIMESTAMP_QUERY
+ = "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =? WHERE
processorName =? AND messageId = ?";
+
private final StampedLock sl = new StampedLock();
private final Set<ProcessorNameAndMessageId> processorNameMessageIdSet =
new HashSet<>();
@@ -62,8 +65,7 @@ public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdReposi
/** intervals after which keep alive is sent for the locks held by an
instance **/
private long lockKeepAliveIntervalMillis;
- private String updateTimestampQuery
- = "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =? WHERE
processorName =? AND messageId = ?";
+ private String updateTimestampQuery = DEFAULT_UPDATE_TIMESTAMP_QUERY;
public JdbcOrphanLockAwareIdempotentRepository(CamelContext camelContext) {
super();
@@ -150,8 +152,15 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
super.doInit();
if (getTableName() != null) {
- updateTimestampQuery =
updateTimestampQuery.replaceFirst(DEFAULT_TABLENAME, getTableName());
+ // update query string from default table name to the new table
name
+ updateTimestampQuery =
DEFAULT_UPDATE_TIMESTAMP_QUERY.replace(DEFAULT_TABLENAME, getTableName());
}
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
executorServiceManager = context.getExecutorServiceManager();
executorService =
executorServiceManager.newSingleThreadScheduledExecutor(this,
this.getClass().getSimpleName());
@@ -162,10 +171,12 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
}
@Override
- protected void doShutdown() throws Exception {
+ protected void doStop() throws Exception {
if (executorServiceManager != null && executorService != null) {
executorServiceManager.shutdownGraceful(executorService);
+ executorService = null;
}
+ super.doStop();
}
@Override
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryRestartTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryRestartTest.java
new file mode 100644
index 000000000000..3f6a4404ea50
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcCachedMessageIdRepositoryRestartTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping and starting the repository again keeps the query that reloads the
cache for a custom table name.
+ */
+public class JdbcCachedMessageIdRepositoryRestartTest {
+
+ private static final String TABLE_NAME = "CAMEL_MESSAGEPROCESSED_ORDERS";
+ private static final String PROCESSOR_NAME = "APP_1";
+ private static final int RESTARTS = 3;
+
+ private EmbeddedDatabase dataSource;
+ private JdbcCachedMessageIdRepository repository;
+
+ @BeforeEach
+ public void setup() {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ repository = new JdbcCachedMessageIdRepository(dataSource,
PROCESSOR_NAME);
+ repository.setTableName(TABLE_NAME);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ repository.shutdown();
+ dataSource.shutdown();
+ }
+
+ @Test
+ public void testRestartKeepsQueryAllString() {
+ String expected = "SELECT messageId, COUNT(*) FROM " + TABLE_NAME + "
WHERE processorName = ? GROUP BY messageId";
+
+ repository.start();
+ assertEquals(expected, repository.getQueryAllString());
+
+ for (int i = 0; i < RESTARTS; i++) {
+ repository.stop();
+ repository.start();
+ }
+ assertEquals(expected, repository.getQueryAllString());
+
+ // a key added by another node while this one is stopped is in the
cache after the next start
+ repository.stop();
+ new JdbcTemplate(dataSource).update(
+ "INSERT INTO " + TABLE_NAME + " (processorName, messageId,
createdAt) VALUES (?, ?, CURRENT_TIMESTAMP)",
+ PROCESSOR_NAME, "FILE_1");
+ repository.start();
+
+ int hits = repository.getHitCount();
+ assertTrue(repository.contains("FILE_1"));
+ assertEquals(hits + 1, repository.getHitCount());
+ }
+}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryRestartTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryRestartTest.java
new file mode 100644
index 000000000000..dbb40986998e
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryRestartTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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 java.sql.Timestamp;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.engine.DefaultExecutorServiceManager;
+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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Stopping and starting the repository again keeps the SQL for a custom table
name and runs one keep-alive task.
+ */
+public class JdbcOrphanLockAwareIdempotentRepositoryRestartTest {
+
+ private static final String TABLE_NAME = "CAMEL_MESSAGEPROCESSED_ORDERS";
+ private static final String PROCESSOR_NAME = "APP_1";
+ private static final long LOCK_MAX_AGE_MILLIS = 300_000L;
+ private static final int RESTARTS = 3;
+
+ private final List<ScheduledExecutorService> keepAliveExecutors = new
CopyOnWriteArrayList<>();
+ private EmbeddedDatabase dataSource;
+ private DefaultCamelContext context;
+ private JdbcOrphanLockAwareIdempotentRepository repository;
+
+ @BeforeEach
+ public void setup() {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+
+ context = new DefaultCamelContext();
+ context.setExecutorServiceManager(new
DefaultExecutorServiceManager(context) {
+ @Override
+ public ScheduledExecutorService
newSingleThreadScheduledExecutor(Object source, String name) {
+ ScheduledExecutorService answer =
super.newSingleThreadScheduledExecutor(source, name);
+ keepAliveExecutors.add(answer);
+ return answer;
+ }
+ });
+
+ repository = new JdbcOrphanLockAwareIdempotentRepository(dataSource,
PROCESSOR_NAME, context);
+ repository.setTableName(TABLE_NAME);
+ repository.setLockMaxAgeMillis(LOCK_MAX_AGE_MILLIS);
+ // the test calls keepAlive() itself
+ repository.setLockKeepAliveIntervalMillis(LOCK_MAX_AGE_MILLIS / 2);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ repository.shutdown();
+ context.stop();
+ dataSource.shutdown();
+ }
+
+ @Test
+ public void testRestartKeepsUpdateTimestampQuery() {
+ String expected = "UPDATE " + TABLE_NAME + " SET createdAt =? WHERE
processorName =? AND messageId = ?";
+
+ repository.start();
+ assertEquals(expected, repository.getUpdateTimestampQuery());
+
+ for (int i = 0; i < RESTARTS; i++) {
+ repository.stop();
+ repository.start();
+ }
+ assertEquals(expected, repository.getUpdateTimestampQuery());
+
+ // the keep-alive still refreshes a lease after the restarts
+ assertTrue(repository.add("FILE_1"));
+ Timestamp expired = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS - 60_000L);
+ new JdbcTemplate(dataSource).update("UPDATE " + TABLE_NAME + " SET
createdAt = ? WHERE messageId = ?", expired,
+ "FILE_1");
+ assertFalse(repository.contains("FILE_1"));
+
+ repository.keepAlive();
+ assertTrue(repository.contains("FILE_1"));
+ }
+
+ @Test
+ public void testRestartUsesOneKeepAliveExecutor() {
+ repository.start();
+ assertEquals(1, keepAliveExecutors.size());
+
+ for (int i = 0; i < RESTARTS; i++) {
+ repository.stop();
+
assertTrue(keepAliveExecutors.stream().allMatch(ExecutorService::isShutdown),
+ "The keep-alive executor should be shut down when the
repository is stopped");
+ repository.start();
+ }
+
+ assertEquals(RESTARTS + 1, keepAliveExecutors.size());
+ assertEquals(1, keepAliveExecutors.stream().filter(e ->
!e.isShutdown()).count());
+
+ repository.shutdown();
+
assertTrue(keepAliveExecutors.stream().allMatch(ExecutorService::isShutdown));
+ }
+}
