This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new dfab1a599990 [backport camel-4.22.x] CAMEL-25778: Fix file consumer
routing a file twice when committed on another thread (#29154)
dfab1a599990 is described below
commit dfab1a599990e3192c6ec8ee178d16a03e5eea17
Author: Guillaume Nodet <[email protected]>
AuthorDate: Wed Nov 4 16:27:54 2026 +0100
[backport camel-4.22.x] CAMEL-25778: Fix file consumer routing a file twice
when committed on another thread (#29154)
---
.../apache/camel/component/file/FileConsumer.java | 30 ++++++
.../file/FileConsumerAsyncCommitRaceTest.java | 116 +++++++++++++++++++++
.../file/FileConsumerCommittedBeforeClaimTest.java | 100 ++++++++++++++++++
3 files changed, 246 insertions(+)
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
index 7d5cd0f30726..98ab584925a2 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/FileConsumer.java
@@ -97,6 +97,36 @@ public class FileConsumer extends GenericFileConsumer<File>
implements ResumeAwa
}
}
+ @Override
+ protected boolean isValidFile(
+ Exchange dynamic,
+ Supplier<GenericFile<File>> file, String name, String
absoluteFilePath,
+ Supplier<String> relativeFilePath, boolean isDirectory, File[]
files) {
+ if (!super.isValidFile(dynamic, file, name, absoluteFilePath,
relativeFilePath, isDirectory, files)) {
+ return false;
+ }
+ if (isDirectory) {
+ return true;
+ }
+ // CAMEL-25778: TOCTOU guard — after super.isValidFile() added the
file to the
+ // in-progress repository, a concurrent async-commit thread may have
already moved
+ // the file away (e.g. to .camel). Use a plain File.exists() check (no
network cost)
+ // to detect this; if the file is gone, undo the in-progress claim and
drop the
+ // eagerly-added idempotent key so the file is not routed a second
time.
+ if (!new File(absoluteFilePath).exists()) {
+ LOG.debug("Skipping file as it no longer exists (already committed
by another thread): {}", name);
+ getEndpoint().getInProgressRepository().remove(absoluteFilePath);
+ // Drop the eager idempotent key so a future file with the same
name is not
+ // permanently blacklisted (mirrors the key resolution in
notUnique()).
+ if (Boolean.TRUE.equals(getEndpoint().isIdempotent()) &&
getEndpoint().isIdempotentEager()
+ && getEndpoint().getIdempotentRepository() != null) {
+ removeExcessiveIdempotentFile(file.get(), dynamic);
+ }
+ return false;
+ }
+ return true;
+ }
+
private boolean pollDirectory(Exchange dynamic, File directory,
List<GenericFile<File>> fileList, int depth) {
depth++;
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerAsyncCommitRaceTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerAsyncCommitRaceTest.java
new file mode 100644
index 000000000000..f116f9611982
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerAsyncCommitRaceTest.java
@@ -0,0 +1,116 @@
+/*
+ * 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;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Regression test for CAMEL-25778: a file consumer can route a file twice
when the exchange is committed on a different
+ * thread.
+ * <p/>
+ * The race: the poll thread lists the directory; another thread commits
(moves the file to .camel and removes it from
+ * the in-progress repository); the poll thread's in-progress check now passes
and routes the already-moved file a
+ * second time. The fix adds a post-add existence check inside {@code
FileConsumer.isValidFile()}.
+ * <p/>
+ * This test is timing-sensitive: reproduction depends on the poll thread and
the commit thread interleaving at the
+ * right moment. It serves as a regression guard that catches the race when
timing cooperates, but may not reproduce it
+ * on every run.
+ */
+class FileConsumerAsyncCommitRaceTest extends ContextTestSupport {
+
+ /** Counts how many times each file name was processed. */
+ private final ConcurrentMap<String, AtomicInteger> processedCount = new
ConcurrentHashMap<>();
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // delay=5ms to poll fast; threads() hands exchange to another
+ // thread, so the commit happens off the poll thread — exactly
the
+ // scenario that triggers the CAMEL-25778 race condition.
+ // maxQueueSize(0) uses a SynchronousQueue: the poll thread
blocks
+ // until a worker picks up the exchange, maximising thread
interleaving.
+ from(fileUri("?sortBy=file:name&delay=5&synchronous=false"))
+ .routeId("myRoute")
+ .autoStartup(false)
+ .threads(1, 10)
+ .maxQueueSize(0)
+ .process(new CountingProcessor(processedCount))
+ .to("mock:done");
+ }
+ };
+ }
+
+ @Test
+ void testNoFilesRoutedTwiceWithAsyncCommit() throws Exception {
+ int fileCount = 50;
+ MockEndpoint mock = getMockEndpoint("mock:done");
+ mock.expectedMessageCount(fileCount);
+ mock.setResultWaitTime(60_000);
+
+ createFiles(fileCount);
+ context.getRouteController().startRoute("myRoute");
+
+ mock.assertIsSatisfied();
+
+ // Verify that no file was processed more than once
+ int duplicates = 0;
+ for (var entry : processedCount.entrySet()) {
+ int count = entry.getValue().get();
+ if (count > 1) {
+ duplicates++;
+ log.warn("File routed {} times: {}", count, entry.getKey());
+ }
+ }
+ assertEquals(0, duplicates, "No file should be routed more than once
(CAMEL-25778)");
+ }
+
+ private void createFiles(int count) throws Exception {
+ Path dir = testDirectory();
+ for (int i = 0; i < count; i++) {
+ Files.writeString(dir.resolve(String.format("file-%04d.txt", i)),
String.format("content-%04d", i));
+ }
+ }
+
+ private static class CountingProcessor implements Processor {
+ private final ConcurrentMap<String, AtomicInteger> processedCount;
+
+ CountingProcessor(ConcurrentMap<String, AtomicInteger> processedCount)
{
+ this.processedCount = processedCount;
+ }
+
+ @Override
+ public void process(Exchange exchange) {
+ String fileName =
exchange.getMessage().getHeader(FileConstants.FILE_NAME, String.class);
+ processedCount.computeIfAbsent(fileName, k -> new
AtomicInteger(0)).incrementAndGet();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommittedBeforeClaimTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommittedBeforeClaimTest.java
new file mode 100644
index 000000000000..5e0332d3a45a
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerCommittedBeforeClaimTest.java
@@ -0,0 +1,100 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Deterministic regression test for CAMEL-25778: verifies {@code
FileConsumer.isValidFile()} correctly handles the case
+ * where a concurrent commit thread moves the file between {@code
inProgressRepository.contains()} and {@code add()}.
+ * <p/>
+ * The race is made deterministic by injecting a custom {@code
inProgressRepository} whose {@code add()} deletes the
+ * file before delegating to the real repository — exactly what a concurrent
commit thread would do (move file + remove
+ * in-progress entry). Without the fix, the file would be routed and the eager
idempotent key would remain. With the
+ * fix, neither happens.
+ */
+class FileConsumerCommittedBeforeClaimTest extends ContextTestSupport {
+
+ private final MemoryIdempotentRepository idempotent = new
MemoryIdempotentRepository();
+ private final AtomicBoolean claimed = new AtomicBoolean();
+
+ /** Custom in-progress repository that simulates a concurrent commit
deleting the file inside add(). */
+ private final MemoryIdempotentRepository inProgress = new
MemoryIdempotentRepository() {
+ @Override
+ public boolean add(String key) {
+ // Simulate: a commit on another thread lands between contains()
and add() —
+ // the file is moved away (here: deleted) before this poll's add()
completes.
+ try {
+ Files.deleteIfExists(Path.of(key));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ claimed.set(true);
+ return super.add(key);
+ }
+ };
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ context.getRegistry().bind("inProgress", inProgress);
+ context.getRegistry().bind("idem", idempotent);
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(fileUri("?initialDelay=0&delay=10"
+ + "&inProgressRepository=#inProgress"
+ + "&idempotent=true&idempotentRepository=#idem"))
+ .routeId("myRoute").autoStartup(false)
+ .to("mock:done");
+ }
+ };
+ }
+
+ @Test
+ void testFileGoneAfterClaimIsSkipped() throws Exception {
+ getMockEndpoint("mock:done").expectedMessageCount(0);
+
+ Path file = testDirectory().resolve("hello.txt");
+ Files.writeString(file, "Hello");
+ String key = file.toAbsolutePath().toString();
+
+ context.getRouteController().startRoute("myRoute");
+
+ // Wait until the in-progress add() was triggered and cleanup has
completed on the poll thread
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertTrue(claimed.get(), "in-progress add() must have been
triggered");
+ assertFalse(inProgress.contains(key), "in-progress key must be
removed after skip");
+ assertFalse(idempotent.contains(key), "eager idempotent key must
be removed after skip");
+ });
+
+ getMockEndpoint("mock:done").assertIsSatisfied(2000);
+ }
+}
