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:30 UTC
(camel) branch fix/CAMEL-24484 created (now 88966862aab7)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24484
in repository https://gitbox.apache.org/repos/asf/camel.git
at 88966862aab7 CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry
This branch includes the following new commits:
new 88966862aab7 CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry
Posted by da...@apache.org.
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");
+ }
+ };
+ }
+}