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 c57570acff35 CAMEL-26816: camel-sql -
JdbcOrphanLockAwareIdempotentRepository keeps a confirmed key after its
instance stops (#29586)
c57570acff35 is described below
commit c57570acff35820e40d10757ac7d6392bfcbe6ff
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 13 03:14:59 2026 +0530
CAMEL-26816: camel-sql - JdbcOrphanLockAwareIdempotentRepository keeps a
confirmed key after its instance stops (#29586)
JdbcOrphanLockAwareIdempotentRepository used the inherited confirm(), so a
processed and confirmed key stayed a lease: it was kept alive while its
instance ran (the keep-alive set grew with every key), and once the instance
stopped another instance took it over and processed the message again.
- confirm() sets createdAt of the key to null (new confirmQuery option; its
default follows tableName) and the key leaves the keep-alive set; it is
kept
alive again when the confirm fails or its transaction does not commit
- the orphan check counts a null createdAt as present and the takeover never
matches it, so a confirmed key stays a duplicate; the keep-alive skips it
- at start, a rolled-back check verifies the confirm query stores null (NOT
NULL columns, MySQL/MariaDB TIMESTAMP with
explicit_defaults_for_timestamp=OFF);
when it does not, one WARN is logged and confirmed keys stay locks as
before
- a confirm that fails at run time returns false and logs a WARN
- confirmQuery set to null keeps the old behaviour
- docs: Orphan Lock section of the sql component and a 4.24 upgrade-guide
entry
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/sql-component.adoc | 11 +-
.../camel-sql/src/main/docs/sql-component.adoc | 11 +-
.../JdbcOrphanLockAwareIdempotentRepository.java | 204 +++++++++-
...anLockAwareIdempotentRepositoryConfirmTest.java | 417 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 20 +
5 files changed, 657 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
index 8bfd8c543df7..5178ca0973b2 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/sql-component.adoc
@@ -754,13 +754,22 @@ One of the limitations of
`org.apache.camel.processor.idempotent.jdbc.JdbcMessag
* lock already exists and the `createdAt` < `System.currentTimeMillis() -
lockMaxAgeMillis`. In this case, it is assumed that there is no active instance
which has the lock and the lock is provided to the requesting instance. The
reason behind is that if the original instance which had the lock, if it was
still running, it would have updated the Timestamp on createdAt using its
keepAlive mechanism
-This repository has two additional configuration parameters
+* lock already exists and its `createdAt` is null: the key was confirmed, so
it is a duplicate, as in `JdbcMessageIdRepository`.
+
+A key that is confirmed (the message was processed, for example by the
Idempotent Consumer, by the file consumer with `idempotent=true` and the
default `idempotentEager=true`, or by the file consumer with
`readLock=idempotent` (also `idempotent-changed` and `idempotent-rename`) and
the default `readLockRemoveOnCommit=false`) is no longer a lock: its
`createdAt` is set to null and it is no longer kept alive, so another instance
does not take it over when the instance that added it stops. [...]
+
+A confirmed key has no age, so a purge by `createdAt` does not delete it. To
purge confirmed keys by age, add a column of your own to the table, for example
`insertedAt TIMESTAMP DEFAULT CURRENT_TIMESTAMP` (the repository does not write
it), or set `confirmQuery` to a statement that also stores the time of the
confirm in such a column, and purge by that column.
+
+The `createdAt` column must store null. The repository checks this once when
it starts, in a transaction that it rolls back: it inserts a row, confirms it
with `confirmQuery`, and checks that its `createdAt` is null. When the check
fails (a `NOT NULL` column, a MySQL or MariaDB `TIMESTAMP` column with
`explicit_defaults_for_timestamp=OFF`, or an error of the database), it logs a
WARN and does not confirm keys until the repository is started again: they stay
locks, as before, and a confir [...]
+
+This repository has these additional configuration parameters
[cols="1,1"]
|===
|Parameter | Description
|`lockMaxAgeMillis` | This refers to the duration after which the lock is
considered orphaned, i.e., if the `currentTimestamp - createdAt >=
lockMaxAgeMillis` then lock is orphaned.
|`lockKeepAliveIntervalMillis` | The frequency at which keep-alive updates are
done to createdAt Timestamp column.
+|`confirmQuery` | The query that confirms a key by setting its `createdAt` to
null; it may set other columns of the table too. It follows `tableName`; set it
when the table is set through custom queries instead. Set it to null or empty
to keep confirmed keys as locks. It is checked when the repository starts; a
query set later is used after the next start.
|===
==== Caching Jdbc IdempotentRepository
diff --git a/components/camel-sql/src/main/docs/sql-component.adoc
b/components/camel-sql/src/main/docs/sql-component.adoc
index 8bfd8c543df7..5178ca0973b2 100644
--- a/components/camel-sql/src/main/docs/sql-component.adoc
+++ b/components/camel-sql/src/main/docs/sql-component.adoc
@@ -754,13 +754,22 @@ One of the limitations of
`org.apache.camel.processor.idempotent.jdbc.JdbcMessag
* lock already exists and the `createdAt` < `System.currentTimeMillis() -
lockMaxAgeMillis`. In this case, it is assumed that there is no active instance
which has the lock and the lock is provided to the requesting instance. The
reason behind is that if the original instance which had the lock, if it was
still running, it would have updated the Timestamp on createdAt using its
keepAlive mechanism
-This repository has two additional configuration parameters
+* lock already exists and its `createdAt` is null: the key was confirmed, so
it is a duplicate, as in `JdbcMessageIdRepository`.
+
+A key that is confirmed (the message was processed, for example by the
Idempotent Consumer, by the file consumer with `idempotent=true` and the
default `idempotentEager=true`, or by the file consumer with
`readLock=idempotent` (also `idempotent-changed` and `idempotent-rename`) and
the default `readLockRemoveOnCommit=false`) is no longer a lock: its
`createdAt` is set to null and it is no longer kept alive, so another instance
does not take it over when the instance that added it stops. [...]
+
+A confirmed key has no age, so a purge by `createdAt` does not delete it. To
purge confirmed keys by age, add a column of your own to the table, for example
`insertedAt TIMESTAMP DEFAULT CURRENT_TIMESTAMP` (the repository does not write
it), or set `confirmQuery` to a statement that also stores the time of the
confirm in such a column, and purge by that column.
+
+The `createdAt` column must store null. The repository checks this once when
it starts, in a transaction that it rolls back: it inserts a row, confirms it
with `confirmQuery`, and checks that its `createdAt` is null. When the check
fails (a `NOT NULL` column, a MySQL or MariaDB `TIMESTAMP` column with
`explicit_defaults_for_timestamp=OFF`, or an error of the database), it logs a
WARN and does not confirm keys until the repository is started again: they stay
locks, as before, and a confir [...]
+
+This repository has these additional configuration parameters
[cols="1,1"]
|===
|Parameter | Description
|`lockMaxAgeMillis` | This refers to the duration after which the lock is
considered orphaned, i.e., if the `currentTimestamp - createdAt >=
lockMaxAgeMillis` then lock is orphaned.
|`lockKeepAliveIntervalMillis` | The frequency at which keep-alive updates are
done to createdAt Timestamp column.
+|`confirmQuery` | The query that confirms a key by setting its `createdAt` to
null; it may set other columns of the table too. It follows `tableName`; set it
when the table is set through custom queries instead. Set it to null or empty
to keep confirmed keys as locks. It is checked when the repository starts; a
query set later is used after the next start.
|===
==== Caching Jdbc IdempotentRepository
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 c7a9d1b741d3..66d9a8514014 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
@@ -20,6 +20,7 @@ import java.sql.Timestamp;
import java.util.List;
import java.util.Objects;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -29,7 +30,12 @@ import javax.sql.DataSource;
import org.apache.camel.CamelContext;
import org.apache.camel.spi.ExecutorServiceManager;
+import org.apache.camel.util.ObjectHelper;
+import org.springframework.dao.DataAccessException;
+import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionException;
import org.springframework.transaction.support.TransactionSynchronization;
import
org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate;
@@ -44,11 +50,26 @@ import
org.springframework.transaction.support.TransactionTemplate;
* A lock is granted to an instance if either the entry for the lock
attributes do not exists in the
* CAMEL_MESSAGEPROCESSED table or if in case the instance holding the lock
has crashed. This is determined if the
* timestamp on the createdAt column is more than the lockMaxAge.
+ *
+ * A key that is confirmed (the message was processed) is no longer a lock:
its createdAt is set to null, which the
+ * orphan check never treats as old, and it is no longer kept alive. So it
stays a duplicate after the instance that
+ * added it stops, as in {@link JdbcMessageIdRepository}. A confirmed key has
no age then, so a purge by createdAt does
+ * not delete it.
+ * <p/>
+ * The createdAt column must store null. The repository checks this once when
it starts, in a transaction that it rolls
+ * back (it inserts a row, confirms it with the confirm query, and checks that
createdAt is null). When the check fails
+ * (a NOT NULL column, or a MySQL or MariaDB TIMESTAMP column with
explicit_defaults_for_timestamp=OFF), it logs a WARN
+ * and does not confirm keys at all: they stay locks, as before, and a confirm
never sends a statement that could fail
+ * inside a transaction of the caller. A confirm that fails later (for example
the connection is lost) logs a WARN, and
+ * when the repository joins a transaction of the caller (same DataSource), it
marks that transaction as rollback only,
+ * as a failed add or remove does.
*/
public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdRepository {
protected static final String DEFAULT_UPDATE_TIMESTAMP_QUERY
= "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =? WHERE
processorName =? AND messageId = ?";
+ protected static final String DEFAULT_CONFIRM_QUERY
+ = "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt = NULL WHERE
processorName = ? AND messageId = ?";
// a concurrent set, so no lock is held while SQL runs: add and remove
hold a connection of the pool while they
// run, and the keep-alive needs one too, so a lock held across SQL could
deadlock with a small pool
@@ -68,6 +89,11 @@ public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdReposi
private String updateTimestampQuery = DEFAULT_UPDATE_TIMESTAMP_QUERY;
+ private String confirmQuery = DEFAULT_CONFIRM_QUERY;
+
+ // set when the repository starts: whether the confirm query stores null
in createdAt
+ private volatile boolean confirmChecked;
+
public JdbcOrphanLockAwareIdempotentRepository(CamelContext camelContext) {
super();
this.context = camelContext;
@@ -97,7 +123,8 @@ public class JdbcOrphanLockAwareIdempotentRepository extends
JdbcMessageIdReposi
* which had acquired the lock has died
*/
// NOTE: the querystring is passed by the user, so, we are safe as he
decide what query wants to perform.
- String orphanLockRecoverQueryString = getQueryString() + " AND
createdAt >= ?";
+ // a confirmed key (createdAt is null) is never an orphan lock
+ String orphanLockRecoverQueryString = getQueryString() + " AND
(createdAt >= ? OR createdAt IS NULL)";
Timestamp xMillisAgo = new Timestamp(System.currentTimeMillis() -
lockMaxAgeMillis);
return jdbcTemplate.queryForObject(orphanLockRecoverQueryString,
Integer.class, processorName, key, // NOSONAR
xMillisAgo);
@@ -106,9 +133,134 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
@Override
protected int delete(String key) {
int result = super.delete(key);
+ stopKeepAlive(key);
+ return result;
+ }
+
+ /**
+ * Confirms the key: the message was processed, so the key is no longer a
lock that is kept alive and that another
+ * instance can take over when this one stops. Its createdAt is set to
null, which the orphan check never treats as
+ * old, so the key stays a duplicate for good.
+ * <p/>
+ * When the key cannot be confirmed (a failure of the database), it stays
a lock that is kept alive while this
+ * instance runs, as before, and false is returned; also when there is no
such key. With no confirm query (null or
+ * empty), or when the check at start found that the confirm query cannot
set createdAt to null, a key is not
+ * confirmed either, no statement is sent, and true is returned. The same
holds before the repository is started,
+ * and after the confirm query is changed, until the next start.
+ *
+ * @return {@code true} if the key was confirmed, {@code false} otherwise
+ */
+ @Override
+ public boolean confirm(final String key) {
+ String query = getConfirmQuery();
+ if (ObjectHelper.isEmpty(query) || !confirmChecked) {
+ // confirmed keys are kept as locks
+ return true;
+ }
+ ProcessorNameAndMessageId id = new
ProcessorNameAndMessageId(processorName, key);
+ // the key leaves the keep-alive right away, as in delete: inside a
transaction of the caller, the confirm locks
+ // the row until that transaction ends, and a later keep-alive that
updated the row would wait for it (and could
+ // deadlock with a later confirm or remove in the same transaction).
It is kept alive again when the confirm fails
+ // or its transaction does not commit; not when no row matched, as
then there is no lease of this instance.
+ boolean keptAlive = processorNameMessageIdSet.remove(id);
+ boolean completed = false;
+ try {
+ // in a transaction of its own (or the one of the caller), as add
and remove
+ Boolean rc = transactionTemplate.execute(status -> {
+ if (jdbcTemplate.update(query, processorName, key) == 0) {
+ return false;
+ }
+ // an insert of the key earlier in this transaction does not
keep it alive after the commit
+ cancelKeepAliveAfterCommit(id);
+ if (keptAlive &&
TransactionSynchronizationManager.isSynchronizationActive()) {
+
TransactionSynchronizationManager.registerSynchronization(new
KeepAliveUnlessCommitted(this, id));
+ }
+ return true;
+ });
+ completed = true;
+ return Boolean.TRUE.equals(rc);
+ } catch (DataAccessException | TransactionException e) {
+ log.warn("Cannot confirm the key {} of processor {} in the
idempotent repository: it stays a lock that is kept"
+ + " alive while this instance runs and that another
instance takes over when this one stops"
+ + " (inside a transaction of the caller, that transaction
is marked as rollback only). Caused by: {}",
+ key, processorName, e.getMessage());
+ log.debug("Cannot confirm the key {} of processor {} in the
idempotent repository", key, processorName, e);
+ return false;
+ } finally {
+ if (keptAlive && !completed) {
+ processorNameMessageIdSet.add(id);
+ }
+ }
+ }
+
+ /**
+ * Checks that the confirm query stores null in createdAt, in a new
transaction that is always rolled back: inserts
+ * a row with a key of its own, confirms it, and reads it back. When the
check fails, keys are not confirmed (they
+ * stay locks, as before), so that a confirm never sends a statement that
fails inside a transaction of the caller.
+ */
+ private boolean checkConfirmQuery(String query) {
+ TransactionTemplate check = new
TransactionTemplate(transactionTemplate.getTransactionManager(),
transactionTemplate);
+
check.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+ String key = UUID.randomUUID().toString();
+ Boolean storesNull;
+ try {
+ storesNull = check.execute(status -> {
+ status.setRollbackOnly();
+ jdbcTemplate.update(getInsertString(), processorName, key, new
Timestamp(System.currentTimeMillis()));
+ int updated;
+ try {
+ updated = jdbcTemplate.update(query, processorName, key);
+ } catch (DataIntegrityViolationException e) {
+ // a NOT NULL column
+ log.debug("The confirm query cannot set createdAt to
null", e);
+ return false;
+ }
+ if (updated == 0) {
+ throw new IllegalStateException("the confirm query updated
no row");
+ }
+ // such as a MySQL or MariaDB TIMESTAMP column with
explicit_defaults_for_timestamp=OFF, which stores the
+ // current time for null
+ Integer stored = jdbcTemplate.queryForObject(getQueryString()
+ " AND createdAt IS NULL", Integer.class,
+ processorName, key);
+ return stored != null && stored > 0;
+ });
+ } catch (RuntimeException e) {
+ log.warn("Cannot check the confirm query of processor {} in the
idempotent repository ({}): keys are not"
+ + " confirmed until the repository is started again. They
stay locks that another instance takes over"
+ + " when this one stops.",
+ processorName, e.getMessage() != null ? e.getMessage() :
e.getClass().getName());
+ log.debug("Cannot check the confirm query of processor {} in the
idempotent repository", processorName, e);
+ return false;
+ } finally {
+ // when the transaction manager does not manage the connection of
the JdbcTemplate (another DataSource), the
+ // rollback does not remove the row of the check
+ try {
+ jdbcTemplate.update(getDeleteString(), processorName, key);
+ } catch (RuntimeException e) {
+ log.debug("Cannot delete the row of the check of the confirm
query", e);
+ }
+ }
+ if (!Boolean.TRUE.equals(storesNull)) {
+ log.warn("Confirmed keys of processor {} stay locks that another
instance takes over when this one stops, as the"
+ + " createdAt column does not store null. Make the
createdAt column nullable (on MySQL or MariaDB with"
+ + " explicit_defaults_for_timestamp=OFF declare it
TIMESTAMP NULL), or set confirmQuery to null or empty"
+ + " to keep confirmed keys as locks.",
+ processorName);
+ return false;
+ }
+ return true;
+ }
+
+ // the key is no longer kept alive, also when an insert of it earlier in
the current transaction would keep it alive
+ // after the commit
+ private void stopKeepAlive(String key) {
ProcessorNameAndMessageId id = new
ProcessorNameAndMessageId(processorName, key);
processorNameMessageIdSet.remove(id);
- // an insert of the same key earlier in this transaction keeps it
alive only after the commit: cancel that
+ cancelKeepAliveAfterCommit(id);
+ }
+
+ // an insert of the key earlier in the current transaction keeps it alive
only after the commit: cancel that
+ private void cancelKeepAliveAfterCommit(ProcessorNameAndMessageId id) {
if (TransactionSynchronizationManager.isSynchronizationActive()) {
for (TransactionSynchronization synchronization :
TransactionSynchronizationManager.getSynchronizations()) {
if (synchronization instanceof KeepAliveAfterCommit pending &&
pending.repository == this
@@ -117,7 +269,6 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
}
}
}
- return result;
}
@Override
@@ -179,6 +330,28 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
}
}
+ /**
+ * Keeps a confirmed key alive again when the transaction of its confirm
does not commit: the row is still a lock
+ * then. When the outcome is unknown, the key is kept alive too: the
keep-alive does not update a row that was
+ * confirmed.
+ */
+ private static final class KeepAliveUnlessCommitted implements
TransactionSynchronization {
+ private final JdbcOrphanLockAwareIdempotentRepository repository;
+ private final ProcessorNameAndMessageId id;
+
+ private
KeepAliveUnlessCommitted(JdbcOrphanLockAwareIdempotentRepository repository,
ProcessorNameAndMessageId id) {
+ this.repository = repository;
+ this.id = id;
+ }
+
+ @Override
+ public void afterCompletion(int status) {
+ if (status != STATUS_COMMITTED) {
+ repository.processorNameMessageIdSet.add(id);
+ }
+ }
+ }
+
@Override
protected void doInit() throws Exception {
if (lockMaxAgeMillis <= lockKeepAliveIntervalMillis) {
@@ -190,6 +363,9 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
if (getTableName() != null) {
// update query string from default table name to the new table
name
updateTimestampQuery =
DEFAULT_UPDATE_TIMESTAMP_QUERY.replace(DEFAULT_TABLENAME, getTableName());
+ if (DEFAULT_CONFIRM_QUERY.equals(confirmQuery)) {
+ confirmQuery =
DEFAULT_CONFIRM_QUERY.replace(DEFAULT_TABLENAME, getTableName());
+ }
}
}
@@ -197,6 +373,9 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
protected void doStart() throws Exception {
super.doStart();
+ String query = getConfirmQuery();
+ confirmChecked = ObjectHelper.isNotEmpty(query) &&
checkConfirmQuery(query);
+
executorServiceManager = context.getExecutorServiceManager();
executorService =
executorServiceManager.newSingleThreadScheduledExecutor(this,
this.getClass().getSimpleName());
@@ -240,7 +419,8 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
.map(processorNameMessageId -> new Object[] {
currentTimestamp,
processorNameMessageId.processorName, processorNameMessageId.messageId })
.collect(Collectors.toList());
- return jdbcTemplate.batchUpdate(getUpdateTimestampQuery(),
args);
+ // not a key that was confirmed meanwhile (its createdAt is
null)
+ return jdbcTemplate.batchUpdate(getUpdateTimestampQuery() + "
AND createdAt IS NOT NULL", args);
});
} catch (Exception e) {
log.error("failed updating createdAt in keepAlive due to ", e);
@@ -259,6 +439,22 @@ public class JdbcOrphanLockAwareIdempotentRepository
extends JdbcMessageIdReposi
this.updateTimestampQuery = updateTimestampQuery;
}
+ public String getConfirmQuery() {
+ return confirmQuery;
+ }
+
+ /**
+ * The query that confirms a key: it sets its createdAt to null. It may
set other columns of a custom table too, for
+ * example the time of the confirm, to purge confirmed keys by age. Set it
to null or empty to keep confirmed keys
+ * as locks that another instance takes over when this one stops (the
behaviour before Camel 4.24). It is checked
+ * when the repository starts: a query set later is not used until the
next start.
+ */
+ public void setConfirmQuery(String confirmQuery) {
+ this.confirmQuery = confirmQuery;
+ // checked again when the repository starts
+ this.confirmChecked = false;
+ }
+
public long getLockMaxAgeMillis() {
return lockMaxAgeMillis;
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryConfirmTest.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryConfirmTest.java
new file mode 100644
index 000000000000..dc8ace80593b
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/idempotent/jdbc/JdbcOrphanLockAwareIdempotentRepositoryConfirmTest.java
@@ -0,0 +1,417 @@
+/*
+ * 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.lang.reflect.Proxy;
+import java.sql.Timestamp;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import
org.apache.camel.processor.idempotent.jdbc.JdbcOrphanLockAwareIdempotentRepository.ProcessorNameAndMessageId;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A key that was processed and confirmed stays a duplicate after the node
that added it stops, while a lease that was
+ * not confirmed is still taken over. Two repository instances (two nodes)
share one H2 database.
+ */
+class JdbcOrphanLockAwareIdempotentRepositoryConfirmTest {
+
+ private static final String PROCESSOR_NAME = "APP_1";
+ private static final long LOCK_MAX_AGE_MILLIS = 300_000L;
+
+ private EmbeddedDatabase dataSource;
+ private JdbcTemplate jdbcTemplate;
+ private CamelContext context;
+ private JdbcOrphanLockAwareIdempotentRepository nodeA;
+ private JdbcOrphanLockAwareIdempotentRepository nodeB;
+
+ @BeforeEach
+ void setup() {
+ dataSource = new EmbeddedDatabaseBuilder()
+ .setType(EmbeddedDatabaseType.H2)
+ .generateUniqueName(true)
+ .build();
+ jdbcTemplate = new JdbcTemplate(dataSource);
+ context = new DefaultCamelContext();
+ nodeA = newRepository();
+ nodeB = newRepository();
+ nodeA.start();
+ nodeB.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ nodeA.shutdown();
+ nodeB.shutdown();
+ context.stop();
+ dataSource.shutdown();
+ }
+
+ @Test
+ void testConfirmedKeyStaysADuplicateAfterItsNodeStops() {
+ assertTrue(nodeA.add("FILE_1"));
+ assertTrue(nodeA.confirm("FILE_1"));
+
+ // node A stops (a rolling restart, or a crash), and the lock max age
passes
+ nodeA.stop();
+ ageLeases();
+
+ assertTrue(nodeB.contains("FILE_1"));
+ assertFalse(nodeB.add("FILE_1"), "A processed and confirmed key must
not be taken over");
+ }
+
+ @Test
+ void testLeaseThatIsNotConfirmedIsTakenOverAfterItsNodeStops() {
+ // control: the orphan lock of an instance that stopped while
processing is still taken over
+ assertTrue(nodeA.add("FILE_2"));
+
+ nodeA.stop();
+ ageLeases();
+
+ assertFalse(nodeB.contains("FILE_2"));
+ assertTrue(nodeB.add("FILE_2"));
+ }
+
+ @Test
+ void testConfirmedKeyIsNoLongerKeptAlive() {
+ assertTrue(nodeA.add("FILE_3"));
+
assertTrue(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_3")));
+
+ assertTrue(nodeA.confirm("FILE_3"));
+
assertFalse(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_3")),
+ "A confirmed key must not be kept alive");
+
+ // a keep-alive does not make the confirmed key a lease again
+ nodeA.keepAlive();
+ assertNull(createdAt("FILE_3"));
+ }
+
+ @Test
+ void testKeyAddedAndConfirmedInOneTransactionIsNotKeptAlive() {
+ TransactionTemplate transaction = new TransactionTemplate(new
DataSourceTransactionManager(dataSource));
+ transaction.executeWithoutResult(status -> {
+ assertTrue(nodeA.add("FILE_4"));
+ assertTrue(nodeA.confirm("FILE_4"));
+ });
+
+
assertFalse(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_4")),
+ "The add of the same transaction must not keep the confirmed
key alive after the commit");
+ nodeA.stop();
+ ageLeases();
+ assertFalse(nodeB.add("FILE_4"));
+ }
+
+ @Test
+ void testConfirmThatRollsBackKeepsTheLeaseAlive() {
+ assertTrue(nodeA.add("FILE_7"));
+
+ TransactionTemplate transaction = new TransactionTemplate(new
DataSourceTransactionManager(dataSource));
+ transaction.executeWithoutResult(status -> {
+ assertTrue(nodeA.confirm("FILE_7"));
+ // the key leaves the keep-alive right away, as in a remove, so a
later keep-alive does not wait for its row
+
assertFalse(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_7")),
+ "A key confirmed in a transaction of the caller must leave
the keep-alive right away");
+ status.setRollbackOnly();
+ });
+
+ // the confirm did not happen, so the lease is kept alive again while
this instance runs
+ assertTrue(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_7")),
+ "A confirm that rolled back must keep the lease alive again");
+ assertNotNull(createdAt("FILE_7"));
+ }
+
+ @Test
+ void testConfirmOfAKeyWhoseRowWasDeletedDoesNotKeepItAlive() {
+ assertTrue(nodeA.add("FILE_15"));
+ // removed meanwhile, for example by a clear of another instance
+ jdbcTemplate.update("DELETE FROM CAMEL_MESSAGEPROCESSED WHERE
messageId = 'FILE_15'");
+
+ assertFalse(nodeA.confirm("FILE_15"));
+
assertFalse(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_15")),
+ "A key without a row must not be kept alive, as another
instance may insert it later");
+ }
+
+ @Test
+ void testKeepAliveDoesNotWaitForAConfirmInACallerTransaction() throws
Exception {
+ assertTrue(nodeA.add("FILE_12"));
+ assertTrue(nodeA.add("FILE_13"));
+ Timestamp old = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS / 2);
+ jdbcTemplate.update("UPDATE CAMEL_MESSAGEPROCESSED SET createdAt = ?
WHERE messageId = 'FILE_13'", old);
+
+ ExecutorService keeper = Executors.newSingleThreadExecutor();
+ try {
+ TransactionTemplate transaction = new TransactionTemplate(new
DataSourceTransactionManager(dataSource));
+ transaction.executeWithoutResult(status -> {
+ // the confirm locks the row of FILE_12 until this transaction
commits
+ assertTrue(nodeA.confirm("FILE_12"));
+ try {
+ // a keep-alive meanwhile refreshes FILE_13, and does not
wait for the row of FILE_12 (which would end
+ // in a lock timeout of the whole batch)
+ keeper.submit(nodeA::keepAlive).get(30, TimeUnit.SECONDS);
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ Timestamp refreshed = createdAt("FILE_13");
+ assertTrue(refreshed.after(old), "The keep-alive must refresh
the other leases while the confirm is open");
+ });
+ } finally {
+ keeper.shutdownNow();
+ }
+ assertNull(createdAt("FILE_12"));
+ }
+
+ @Test
+ void testCheckAtStartWithAConfirmQueryThatDoesNotStoreNull() {
+ // as a MySQL TIMESTAMP column with
explicit_defaults_for_timestamp=OFF, which stores the current time for null
+ List<String> warnings = new CopyOnWriteArrayList<>();
+ JdbcOrphanLockAwareIdempotentRepository node = newRepository();
+ node.log = recordingLogger(warnings);
+ node.setConfirmQuery(
+ "UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =
CURRENT_TIMESTAMP WHERE processorName = ? AND messageId = ?");
+ node.start();
+ try {
+ assertEquals(1, warnings.size(), "The check at start must log one
WARN: " + warnings);
+ assertTrue(node.add("FILE_8"));
+ Timestamp leased = new Timestamp(System.currentTimeMillis() -
60_000L);
+ jdbcTemplate.update("UPDATE CAMEL_MESSAGEPROCESSED SET createdAt =
? WHERE messageId = 'FILE_8'", leased);
+
+ // the key is not confirmed, and no statement is sent: the key
stays a lock that is kept alive, as before
+ assertTrue(node.confirm("FILE_8"));
+
assertTrue(node.getProcessorNameMessageIdSet().contains(id("FILE_8")),
+ "A key that is not confirmed must still be kept alive");
+ assertEquals(leased, createdAt("FILE_8"), "No confirm statement
must be sent");
+ assertEquals(1, warnings.size(), "A confirm must not log again: "
+ warnings);
+ } finally {
+ node.shutdown();
+ }
+ }
+
+ @Test
+ void
testCheckAtStartWithANotNullColumnKeepsTheCallerTransactionCommittable() {
+ jdbcTemplate.execute("CREATE TABLE CAMEL_MESSAGEPROCESSED_NN
(processorName VARCHAR(255), messageId VARCHAR(100), "
+ + "createdAt TIMESTAMP NOT NULL, PRIMARY KEY
(processorName, messageId))");
+ List<String> warnings = new CopyOnWriteArrayList<>();
+ JdbcOrphanLockAwareIdempotentRepository node = newRepository();
+ node.log = recordingLogger(warnings);
+ node.setTableName("CAMEL_MESSAGEPROCESSED_NN");
+ node.start();
+ try {
+ assertEquals(1, warnings.size(), "The check at start must log one
WARN: " + warnings);
+ assertEquals(0, jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
CAMEL_MESSAGEPROCESSED_NN", Integer.class),
+ "The check at start must roll back its row");
+
+ // a transacted route with completionEager=true on the same
DataSource: the confirm must not mark the
+ // transaction of the caller as rollback only, so that it commits
as before
+ TransactionTemplate transaction = new TransactionTemplate(new
DataSourceTransactionManager(dataSource));
+ transaction.executeWithoutResult(status -> {
+ assertTrue(node.add("FILE_9"));
+ assertTrue(node.confirm("FILE_9"));
+ });
+
+ assertEquals(1, jdbcTemplate.queryForObject("SELECT COUNT(*) FROM
CAMEL_MESSAGEPROCESSED_NN", Integer.class));
+
assertTrue(node.getProcessorNameMessageIdSet().contains(id("FILE_9")),
+ "A key that is not confirmed must still be kept alive");
+ } finally {
+ node.shutdown();
+ }
+ }
+
+ @Test
+ void testConfirmThatFailsLogsAWarnEachTime() {
+ List<String> warnings = new CopyOnWriteArrayList<>();
+ nodeA.log = recordingLogger(warnings);
+ assertTrue(nodeA.add("FILE_9"));
+ assertTrue(nodeA.add("FILE_14"));
+ // the column no longer stores null after the start (a failure the
check at start cannot see)
+ jdbcTemplate.execute("ALTER TABLE CAMEL_MESSAGEPROCESSED ALTER COLUMN
createdAt SET NOT NULL");
+
+ assertFalse(nodeA.confirm("FILE_9"));
+ assertFalse(nodeA.confirm("FILE_14"));
+
+ assertTrue(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_9")),
+ "A key that could not be confirmed must still be kept alive");
+
assertTrue(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_14")));
+ assertTrue(nodeA.contains("FILE_9"));
+ assertEquals(2, warnings.size(), "Each failed confirm must log a WARN:
" + warnings);
+ assertFalse(warnings.get(1).contains("nullable"), "A failed confirm
must not log the hint of the check at start");
+ }
+
+ @Test
+ void testNoConfirmQueryKeepsConfirmedKeysAsLocks() {
+ // the way back to the behaviour before: a confirmed key stays a lock
+ nodeA.setConfirmQuery(null);
+ assertTrue(nodeA.add("FILE_10"));
+ assertTrue(nodeA.confirm("FILE_10"));
+
assertTrue(nodeA.getProcessorNameMessageIdSet().contains(id("FILE_10")));
+
+ nodeA.stop();
+ ageLeases();
+ assertTrue(nodeB.add("FILE_10"));
+ }
+
+ @Test
+ void testConfirmWithACustomTableName() {
+ JdbcOrphanLockAwareIdempotentRepository ordersA = newRepository();
+ JdbcOrphanLockAwareIdempotentRepository ordersB = newRepository();
+ ordersA.setTableName("CAMEL_MESSAGEPROCESSED_ORDERS");
+ ordersB.setTableName("CAMEL_MESSAGEPROCESSED_ORDERS");
+ ordersA.start();
+ ordersB.start();
+ try {
+ assertTrue(ordersA.add("ORDER_1"));
+ assertTrue(ordersA.confirm("ORDER_1"));
+
+ ordersA.stop();
+ jdbcTemplate.update("UPDATE CAMEL_MESSAGEPROCESSED_ORDERS SET
createdAt = ? WHERE createdAt IS NOT NULL",
+ new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS - 60_000L));
+ assertFalse(ordersB.add("ORDER_1"));
+ } finally {
+ ordersA.shutdown();
+ ordersB.shutdown();
+ }
+ }
+
+ @Test
+ void testConfirmOfAnUnknownKey() {
+ assertFalse(nodeA.confirm("FILE_5"));
+ assertTrue(nodeA.add("FILE_5"));
+ }
+
+ @Test
+ void testRemovedKeyCanBeAddedAgain() {
+ assertTrue(nodeA.add("FILE_6"));
+ assertTrue(nodeA.confirm("FILE_6"));
+ assertTrue(nodeA.remove("FILE_6"));
+
+ assertTrue(nodeB.add("FILE_6"));
+ }
+
+ @Test
+ void
testIdempotentConsumerSkipsADuplicateAfterTheNodeThatProcessedItStopped()
throws Exception {
+ CamelContext a = new DefaultCamelContext();
+ CamelContext b = new DefaultCamelContext();
+ try {
+ a.addRoutes(route(nodeA));
+ b.addRoutes(route(nodeB));
+ a.start();
+ b.start();
+
+ MockEndpoint processedOnA = a.getEndpoint("mock:processed",
MockEndpoint.class);
+ processedOnA.expectedMessageCount(1);
+ a.createProducerTemplate().sendBodyAndHeader("direct:start",
"order", "orderId", "ORDER_1");
+ processedOnA.assertIsSatisfied();
+
+ // node A stops, the lock max age passes, and a duplicate of the
order reaches node B
+ a.stop();
+ nodeA.stop();
+ ageLeases();
+
+ MockEndpoint processedOnB = b.getEndpoint("mock:processed",
MockEndpoint.class);
+ processedOnB.expectedMessageCount(0);
+ b.createProducerTemplate().sendBodyAndHeader("direct:start",
"order", "orderId", "ORDER_1");
+ processedOnB.assertIsSatisfied();
+ } finally {
+ a.stop();
+ b.stop();
+ }
+ }
+
+ private static RouteBuilder route(JdbcOrphanLockAwareIdempotentRepository
repository) {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .idempotentConsumer(header("orderId"), repository)
+ .to("mock:processed");
+ }
+ };
+ }
+
+ private JdbcOrphanLockAwareIdempotentRepository newRepository() {
+ JdbcOrphanLockAwareIdempotentRepository repository
+ = new JdbcOrphanLockAwareIdempotentRepository(dataSource,
PROCESSOR_NAME, context);
+ repository.setLockMaxAgeMillis(LOCK_MAX_AGE_MILLIS);
+ // the test calls keepAlive() itself
+ repository.setLockKeepAliveIntervalMillis(LOCK_MAX_AGE_MILLIS / 2);
+ return repository;
+ }
+
+ // the time passes: every lease is older than the lock max age, as no node
refreshes it any more
+ private void ageLeases() {
+ Timestamp expired = new Timestamp(System.currentTimeMillis() -
LOCK_MAX_AGE_MILLIS - 60_000L);
+ jdbcTemplate.update("UPDATE CAMEL_MESSAGEPROCESSED SET createdAt = ?
WHERE createdAt IS NOT NULL", expired);
+ }
+
+ private Timestamp createdAt(String key) {
+ return jdbcTemplate.queryForObject(
+ "SELECT createdAt FROM CAMEL_MESSAGEPROCESSED WHERE
processorName = ? AND messageId = ?", Timestamp.class,
+ PROCESSOR_NAME, key);
+ }
+
+ // records the WARN messages of the repository
+ private static Logger recordingLogger(List<String> warnings) {
+ return (Logger) Proxy.newProxyInstance(Logger.class.getClassLoader(),
new Class<?>[] { Logger.class },
+ (proxy, method, args) -> {
+ switch (method.getName()) {
+ case "hashCode":
+ return System.identityHashCode(proxy);
+ case "equals":
+ return proxy == args[0];
+ case "toString":
+ return "recordingLogger";
+ default:
+ break;
+ }
+ if ("warn".equals(method.getName()) && args != null &&
args.length > 0) {
+ warnings.add(String.valueOf(args[0]));
+ }
+ if (method.getReturnType() == boolean.class) {
+ return false;
+ }
+ if (method.getReturnType() == String.class) {
+ return "test";
+ }
+ return null;
+ });
+ }
+
+ private static ProcessorNameAndMessageId id(String key) {
+ return new ProcessorNameAndMessageId(PROCESSOR_NAME, key);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index da6bbccf6431..14a1ddf8a885 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -3229,6 +3229,26 @@ parameter than the statement has placeholders: most JDBC
drivers then fail with
ignore the extra parameter, so the statement still matches an expired entry.
With `tableName` the statements are
derived from the default statements, as before.
+=== camel-sql - JdbcOrphanLockAwareIdempotentRepository keeps confirmed keys
+
+`JdbcOrphanLockAwareIdempotentRepository` now confirms a key by setting its
`createdAt` to null and no longer keeping
+it alive, so a confirmed key stays a duplicate after its instance stops, as in
`JdbcMessageIdRepository`. Before,
+another instance took it over `lockMaxAgeMillis` later and processed the
message again. Keys that are only added
+(the file consumer with `idempotentEager=false`) and keys written before the
upgrade are still taken over. With
+`readLockRemoveOnRollback=false`, a failed file is confirmed too and is no
longer retried by another instance.
+
+The `createdAt` column must allow null. The repository checks this when it
starts (in a transaction that it rolls
+back); with a `NOT NULL` column, or a MySQL/MariaDB `TIMESTAMP` with
`explicit_defaults_for_timestamp=OFF`, it logs a
+WARN and keys stay locks as before, and a confirm sends no statement.
Confirmed keys have no age, so housekeeping by
+`createdAt` does not delete them: to purge them by age, add a column of your
own (such as
+`insertedAt TIMESTAMP DEFAULT CURRENT_TIMESTAMP`) or set a `confirmQuery` that
also stores the time of the confirm,
+and purge by that column. With the `idempotent` read locks of the file
consumer and the default
+`readLockRemoveOnCommit=false`, the name of every processed file stays in the
table; when the key is only a read
+lock, set `readLockRemoveOnCommit=true` with `readLockIdempotentReleaseDelay`.
With a custom table set by custom
+queries instead of `tableName`, set `confirmQuery` too; set it to null or
empty to keep the old behaviour. All
+instances sharing the table must run this version: older ones see a confirmed
key as absent (before 4.14.6 and 4.18.2
+they take it over).
+
=== camel-jta - an exchange marked rollback only last rolls back its
transaction
In a `transacted()` route that uses a JTA transaction manager,
`markRollbackOnlyLast()` now rolls back the
