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 1f35d963cd79 CAMEL-24484: camel-file - readLockRemoveOnCommit=false
must not remove a pre-existing idempotent entry
1f35d963cd79 is described below
commit 1f35d963cd79366d6b102cf1cadb90ce50c711d3
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Aug 26 18:11:08 2026 +0200
CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a
pre-existing idempotent entry
Fixes a regression where readLockRemoveOnCommit=false (and
readLockRemoveOnRollback=false)
was silently ignored: a previously committed idempotent-repository entry
got removed the
next time a file with the same idempotent key was polled.
Two places assumed "file did not start processing" implies "we added this
idempotent entry
ourselves, so it's safe to remove" - false whenever the entry pre-existed.
Fixed by tracking
lock ownership explicitly via a new Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED
property, and by
guarding GenericFileConsumer's "not started" cleanup with isIdempotent().
Co-authored-by: Claude Opus 4.8 <[email protected]>
Co-Authored-By: Claude Sonnet 5 <[email protected]>
Closes #25726
---
.../camel/component/file/GenericFileConsumer.java | 6 +-
...dempotentChangedRepositoryReadLockStrategy.java | 25 ++++-
...IdempotentRenameRepositoryReadLockStrategy.java | 25 ++++-
.../FileIdempotentRepositoryReadLockStrategy.java | 24 ++++-
.../org/apache/camel/ExchangeConstantProvider.java | 3 +-
.../src/main/java/org/apache/camel/Exchange.java | 1 +
...dempotentChangedReadLockRemoveOnCommitTest.java | 82 ++++++++++++++++
...mpotentRepositoryReadLockStrategyAbortTest.java | 105 +++++++++++++++++++++
8 files changed, 262 insertions(+), 9 deletions(-)
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
index fe72a23124b1..f42276871293 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java
@@ -284,7 +284,8 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
String key = file.getAbsoluteFilePath();
endpoint.getInProgressRepository().remove(key);
// if we added eager to idempotent then we need to remove this
- if (endpoint.isIdempotentEager() &&
endpoint.getIdempotentRepository() != null) {
+ if (Boolean.TRUE.equals(endpoint.isIdempotent()) &&
endpoint.isIdempotentEager()
+ && endpoint.getIdempotentRepository() != null) {
removeExcessiveIdempotentFile(file, null);
}
releaseExchange(exchange, true);
@@ -311,7 +312,8 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
for (GenericFile file : files) {
String key = file.getAbsoluteFilePath();
endpoint.getInProgressRepository().remove(key);
- if (endpoint.isIdempotentEager() &&
endpoint.getIdempotentRepository() != null) {
+ if (Boolean.TRUE.equals(endpoint.isIdempotent()) &&
endpoint.isIdempotentEager()
+ && endpoint.getIdempotentRepository() != null) {
removeExcessiveIdempotentFile(file, null);
}
}
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java
index 945277336efc..ce5e6c8c355a 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java
@@ -108,6 +108,10 @@ public class
FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp
idempotentRepository.remove(exchange, key);
}
}
+ // remember whether we own the key end-to-end (idempotent add +
changed check both
+ // succeeded), so releaseExclusiveReadLockOnAbort knows whether it is
safe to remove it if
+ // begin() later fails for an unrelated reason
+ exchange.setProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer);
return answer;
}
@@ -115,8 +119,16 @@ public class
FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp
public void releaseExclusiveReadLockOnAbort(
GenericFileOperations<File> operations, GenericFile<File> file,
Exchange exchange)
throws Exception {
- String key = asKey(exchange, file);
- idempotentRepository.remove(exchange, key);
+ // only remove the key if we own it end-to-end. If we never owned it
(pre-existing key) or
+ // the changed-check failed (already cleaned up above in
acquireExclusiveReadLock) there is
+ // nothing to do here. If we do own it, acquireExclusiveReadLock
succeeded and something
+ // later in begin() must have failed, so the key is ours to clean up
+ boolean acquired
+ = exchange.getProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class);
+ if (acquired) {
+ String key = asKey(exchange, file);
+ idempotentRepository.remove(exchange, key);
+ }
changed.releaseExclusiveReadLockOnAbort(operations, file, exchange);
}
@@ -328,6 +340,15 @@ public class
FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp
return key;
}
+ private static String asReadLockKey(GenericFile<File> file, String key) {
+ // use the copy from absolute path as that was the original path of the
+ // file when the lock was acquired; e.g. if the file consumer uses
preMove
+ // then the file is moved and would otherwise no longer match
+ String path = file.getCopyFromAbsoluteFilePath() != null
+ ? file.getCopyFromAbsoluteFilePath() :
file.getAbsoluteFilePath();
+ return path + "-" + key;
+ }
+
@Override
protected void doStart() throws Exception {
ObjectHelper.notNull(camelContext, "camelContext", this);
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java
index 72ca3c2a866a..0f205dc20bf8 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java
@@ -100,6 +100,10 @@ public class
FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo
idempotentRepository.remove(exchange, key);
}
}
+ // remember whether we own the key end-to-end (idempotent add + rename
check both
+ // succeeded), so releaseExclusiveReadLockOnAbort knows whether it is
safe to remove it if
+ // begin() later fails for an unrelated reason
+ exchange.setProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer);
return answer;
}
@@ -107,8 +111,16 @@ public class
FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo
public void releaseExclusiveReadLockOnAbort(
GenericFileOperations<File> operations, GenericFile<File> file,
Exchange exchange)
throws Exception {
- String key = asKey(exchange, file);
- idempotentRepository.remove(exchange, key);
+ // only remove the key if we own it end-to-end. If we never owned it
(pre-existing key) or
+ // the rename-check failed (already cleaned up above in
acquireExclusiveReadLock) there is
+ // nothing to do here. If we do own it, acquireExclusiveReadLock
succeeded and something
+ // later in begin() must have failed, so the key is ours to clean up
+ boolean acquired
+ = exchange.getProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class);
+ if (acquired) {
+ String key = asKey(exchange, file);
+ idempotentRepository.remove(exchange, key);
+ }
rename.releaseExclusiveReadLockOnAbort(operations, file, exchange);
}
@@ -242,6 +254,15 @@ public class
FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo
return key;
}
+ private static String asReadLockKey(GenericFile<File> file, String key) {
+ // use the copy from absolute path as that was the original path of the
+ // file when the lock was acquired; e.g. if the file consumer uses
preMove
+ // then the file is moved and would otherwise no longer match
+ String path = file.getCopyFromAbsoluteFilePath() != null
+ ? file.getCopyFromAbsoluteFilePath() :
file.getAbsoluteFilePath();
+ return path + "-" + key;
+ }
+
@Override
protected void doStart() throws Exception {
ObjectHelper.notNull(camelContext, "camelContext", this);
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java
index 94c124619af2..1883051b947d 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java
@@ -88,6 +88,9 @@ public class FileIdempotentRepositoryReadLockStrategy extends
ServiceSupport
// another node is processing the file so skip
CamelLogger.log(LOG, readLockLoggingLevel, "Cannot acquire read
lock. Will skip the file: " + file);
}
+ // remember whether we added the key ourselves, so
releaseExclusiveReadLockOnAbort knows
+ // whether it is safe to remove it if begin() later fails for an
unrelated reason
+ exchange.setProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer);
return answer;
}
@@ -95,8 +98,16 @@ public class FileIdempotentRepositoryReadLockStrategy
extends ServiceSupport
public void releaseExclusiveReadLockOnAbort(
GenericFileOperations<File> operations, GenericFile<File> file,
Exchange exchange)
throws Exception {
- String key = asKey(exchange, file);
- idempotentRepository.remove(exchange, key);
+ // only remove the key if we added it during acquireExclusiveReadLock.
If the key already
+ // existed (owned by a previous run or another node) it must not be
touched here. If we did
+ // add it, acquireExclusiveReadLock succeeded and something later in
begin() must have
+ // failed, so the key is ours to clean up
+ boolean acquired
+ = exchange.getProperty(asReadLockKey(file,
Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class);
+ if (acquired) {
+ String key = asKey(exchange, file);
+ idempotentRepository.remove(exchange, key);
+ }
}
@Override
@@ -290,6 +301,15 @@ public class FileIdempotentRepositoryReadLockStrategy
extends ServiceSupport
return key;
}
+ private static String asReadLockKey(GenericFile<File> file, String key) {
+ // use the copy from absolute path as that was the original path of the
+ // file when the lock was acquired; e.g. if the file consumer uses
preMove
+ // then the file is moved and would otherwise no longer match
+ String path = file.getCopyFromAbsoluteFilePath() != null
+ ? file.getCopyFromAbsoluteFilePath() :
file.getAbsoluteFilePath();
+ return path + "-" + key;
+ }
+
@Override
protected void doStart() throws Exception {
ObjectHelper.notNull(camelContext, "camelContext", this);
diff --git
a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
index ec7e00bb731b..9b93e7bb3081 100644
---
a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
+++
b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java
@@ -17,7 +17,7 @@ public class ExchangeConstantProvider {
private static final Map<String, String> MAP;
static {
- Map<String, String> map = new HashMap<>(155);
+ Map<String, String> map = new HashMap<>(156);
map.put("ACTIVITY_SPAN_TAGS", "CamelActivitySpanTags");
map.put("AGGREGATED_COLLECTION_GUARD",
"CamelAggregatedCollectionGuard");
map.put("AGGREGATED_COMPLETED_BY", "CamelAggregatedCompletedBy");
@@ -73,6 +73,7 @@ public class ExchangeConstantProvider {
map.put("FILE_LOCK_EXCLUSIVE_LOCK", "CamelFileLockExclusiveLock");
map.put("FILE_LOCK_FILE_ACQUIRED", "CamelFileLockFileAcquired");
map.put("FILE_LOCK_FILE_NAME", "CamelFileLockFileName");
+ map.put("FILE_LOCK_IDEMPOTENT_ACQUIRED",
"CamelFileLockIdempotentAcquired");
map.put("FILE_LOCK_RANDOM_ACCESS_FILE",
"CamelFileLockRandomAccessFile");
map.put("FILE_NAME", "CamelFileName");
map.put("FILE_NAME_CONSUMED", "CamelFileNameConsumed");
diff --git a/core/camel-api/src/main/java/org/apache/camel/Exchange.java
b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
index d2b64260536d..74457a08e0e8 100644
--- a/core/camel-api/src/main/java/org/apache/camel/Exchange.java
+++ b/core/camel-api/src/main/java/org/apache/camel/Exchange.java
@@ -156,6 +156,7 @@ public interface Exchange extends VariableAware {
String FILE_LENGTH = "CamelFileLength";
String FILE_LOCK_FILE_ACQUIRED = "CamelFileLockFileAcquired";
String FILE_LOCK_FILE_NAME = "CamelFileLockFileName";
+ String FILE_LOCK_IDEMPOTENT_ACQUIRED = "CamelFileLockIdempotentAcquired";
String FILE_LOCK_EXCLUSIVE_LOCK = "CamelFileLockExclusiveLock";
String FILE_LOCK_RANDOM_ACCESS_FILE = "CamelFileLockRandomAccessFile";
String FILE_LOCK_CHANNEL_FILE = "CamelFileLockChannelFile";
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.java
new file mode 100644
index 000000000000..1d35ca58c1cf
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.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.component.file.strategy;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.Registry;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+class FileIdempotentChangedReadLockRemoveOnCommitTest extends
ContextTestSupport {
+
+ final MemoryIdempotentRepository myRepo = new MemoryIdempotentRepository();
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry jndi = super.createCamelRegistry();
+ jndi.bind("myRepo", myRepo);
+ return jndi;
+ }
+
+ @Test
+ void testExistingEntryNotRemovedOnCommitFalse() throws Exception {
+ assertThat(myRepo.getCacheSize()).isZero();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ // drop hello.txt -> processed once, moved to .camel, entry retained
+ template.sendBodyAndHeader(fileUri(), "Hello World",
Exchange.FILE_NAME, "hello.txt");
+ mock.assertIsSatisfied();
+
+ await().atMost(5, TimeUnit.SECONDS).untilAsserted(() ->
assertThat(myRepo.contains("hello.txt")).isTrue());
+
+ // re-drop a file with the same name -> must be skipped as a duplicate
+ // and the existing entry must be retained because
readLockRemoveOnCommit=false
+ mock.reset();
+ mock.expectedMessageCount(0);
+ template.sendBodyAndHeader(fileUri(), "Hello World Again",
Exchange.FILE_NAME, "hello.txt");
+
+ // give the consumer several poll cycles to (wrongly) process/remove it
+ mock.assertIsSatisfied(2000);
+
+ assertThat(myRepo.contains("hello.txt"))
+ .as("existing idempotent entry must be retained when
readLockRemoveOnCommit=false")
+ .isTrue();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("?initialDelay=0&delay=50&readLockCheckInterval=50"
+ +
"&readLock=idempotent-changed&idempotentRepository=#myRepo"
+ +
"&idempotentKey=${file:onlyname}&readLockRemoveOnCommit=false"))
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java
new file mode 100644
index 000000000000..adeb287c4ada
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java
@@ -0,0 +1,105 @@
+/*
+ * 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.file.strategy;
+
+import java.io.File;
+import java.nio.file.Files;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.file.FileEndpoint;
+import org.apache.camel.component.file.GenericFile;
+import org.apache.camel.support.DefaultExchange;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.apache.camel.support.service.ServiceHelper;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Unit tests releaseExclusiveReadLockOnAbort directly: it must only remove
the idempotent key when this strategy itself
+ * added it during acquireExclusiveReadLock. abort() can also be reached after
a successful acquire, e.g. when a preMove
+ * rename fails afterwards, so a pre-existing key (owned by a previous run or
another node) must never be touched, while
+ * a key we own must be cleaned up.
+ */
+class FileIdempotentRepositoryReadLockStrategyAbortTest extends
ContextTestSupport {
+
+ private FileIdempotentRepositoryReadLockStrategy
newStrategy(MemoryIdempotentRepository repo) throws Exception {
+ ServiceHelper.startService(repo);
+
+ FileEndpoint endpoint = context.getEndpoint(fileUri(),
FileEndpoint.class);
+
+ FileIdempotentRepositoryReadLockStrategy strategy = new
FileIdempotentRepositoryReadLockStrategy();
+ strategy.setCamelContext(context);
+ strategy.setIdempotentRepository(repo);
+ strategy.prepareOnStartup(null, endpoint);
+ ServiceHelper.startService(strategy);
+ return strategy;
+ }
+
+ private GenericFile<File> newGenericFile(String name) throws Exception {
+ File file = testFile(name).toFile();
+ Files.createDirectories(file.getParentFile().toPath());
+ Files.writeString(file.toPath(), "Hello World");
+
+ GenericFile<File> genericFile = new GenericFile<>();
+ genericFile.setFile(file);
+ genericFile.setAbsoluteFilePath(file.getAbsolutePath());
+ return genericFile;
+ }
+
+ @Test
+ void testAbortDoesNotRemovePreExistingKey() throws Exception {
+ MemoryIdempotentRepository repo = new MemoryIdempotentRepository();
+ FileIdempotentRepositoryReadLockStrategy strategy = newStrategy(repo);
+
+ GenericFile<File> genericFile = newGenericFile("hello.txt");
+ Exchange exchange = new DefaultExchange(context);
+
+ // simulate the key already being owned by a previous,
already-committed run
+ repo.add(exchange, genericFile.getAbsoluteFilePath());
+
+ boolean acquired = strategy.acquireExclusiveReadLock(null,
genericFile, exchange);
+ assertThat(acquired).as("must not acquire a key that already
exists").isFalse();
+
+ strategy.releaseExclusiveReadLockOnAbort(null, genericFile, exchange);
+
+ assertThat(repo.contains(genericFile.getAbsoluteFilePath()))
+ .as("pre-existing key must be retained since we never owned
it")
+ .isTrue();
+ }
+
+ @Test
+ void testAbortRemovesKeyWeAcquired() throws Exception {
+ MemoryIdempotentRepository repo = new MemoryIdempotentRepository();
+ FileIdempotentRepositoryReadLockStrategy strategy = newStrategy(repo);
+
+ GenericFile<File> genericFile = newGenericFile("hello2.txt");
+ Exchange exchange = new DefaultExchange(context);
+
+ boolean acquired = strategy.acquireExclusiveReadLock(null,
genericFile, exchange);
+ assertThat(acquired).as("must acquire a key that does not yet
exist").isTrue();
+
+ // simulate begin() throwing afterwards for an unrelated reason (e.g.
a preMove rename
+ // failing), which still routes into releaseExclusiveReadLockOnAbort
+ strategy.releaseExclusiveReadLockOnAbort(null, genericFile, exchange);
+
+ assertThat(repo.contains(genericFile.getAbsoluteFilePath()))
+ .as("key we acquired ourselves must be removed on abort")
+ .isFalse();
+ }
+}
