You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/08/25 14:39:31 UTC
(camel) 01/01: CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch fix/CAMEL-24484
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 88966862aab7802d7941556f9dbb65bd42ceb65f
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Aug 25 16:39:11 2026 +0200
CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry
Two independent places assumed that a file which did not start processing
must have been added to the idempotent repository by the current consumer,
so it was safe to remove:
- FileIdempotentRepositoryReadLockStrategy/-Changed/-Rename unconditionally
removed the key in releaseExclusiveReadLockOnAbort(), even when
acquireExclusiveReadLock() failed because the key already existed from a
prior, already committed run. This ignored readLockRemoveOnCommit=false
entirely.
- GenericFileConsumer's "not started" cleanup (CAMEL-21947) removed from
the idempotent repository based on isIdempotentEager() alone, without
checking isIdempotent(), so it also reached into entries populated by an
idempotent-changed/-rename read lock rather than the eager dedup it was
meant to guard.
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../camel/component/file/GenericFileConsumer.java | 6 +-
...dempotentChangedRepositoryReadLockStrategy.java | 6 +-
...IdempotentRenameRepositoryReadLockStrategy.java | 6 +-
.../FileIdempotentRepositoryReadLockStrategy.java | 5 +-
...dempotentChangedReadLockRemoveOnCommitTest.java | 82 ++++++++++++++++++++++
5 files changed, 97 insertions(+), 8 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 76a775a91b74..b192f487f852 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..32774ed9b83b 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
@@ -115,8 +115,10 @@ 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);
+ // this is only called after acquireExclusiveReadLock failed. If it failed because the key
+ // already existed (owned by a previous run or another node) it must not be removed here. If
+ // it failed because the changed-check did not pass after we added the key, acquireExclusiveReadLock
+ // already removed it, so there is nothing left to clean up
changed.releaseExclusiveReadLockOnAbort(operations, file, exchange);
}
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..f40b500b4908 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
@@ -107,8 +107,10 @@ 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);
+ // this is only called after acquireExclusiveReadLock failed. If it failed because the key
+ // already existed (owned by a previous run or another node) it must not be removed here. If
+ // it failed because the rename-check did not pass after we added the key, acquireExclusiveReadLock
+ // already removed it, so there is nothing left to clean up
rename.releaseExclusiveReadLockOnAbort(operations, file, exchange);
}
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..1e65b3423165 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
@@ -95,8 +95,9 @@ 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);
+ // this is only called after acquireExclusiveReadLock failed to add the key, which means
+ // either the key already existed (owned by a previous run or another node) and must not
+ // be touched, so there is nothing to release/remove here
}
@Override
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..1c20cd378028
--- /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;
+
+public 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
+ public 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");
+ }
+ };
+ }
+}