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 7836b5102c1f CAMEL-25960: camel-file - keep the done file of a poll
for its last file that completes asynchronously (#29290)
7836b5102c1f is described below
commit 7836b5102c1fdc12c56e7a51e187a8a2d9e811cc
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Nov 6 12:50:38 2026 +0530
CAMEL-25960: camel-file - keep the done file of a poll for its last file
that completes asynchronously (#29290)
Each file now holds a view of the kept done files of its own poll, so the
last file of a poll that completes asynchronously no longer deletes (or keeps)
the done file depending on when the next poll starts.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/file/GenericFileConsumer.java | 5 +-
.../component/file/GenericFileKeptDoneFiles.java | 53 ++++++--
.../component/file/GenericFileOnCompletion.java | 8 +-
.../file/GenericFileKeptDoneFilesTest.java | 22 ++++
.../FileConsumerDoneFileNameAsyncLastFileTest.java | 134 +++++++++++++++++++++
5 files changed, 209 insertions(+), 13 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 72827db5a9bb..c36d7e223ec4 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
@@ -604,7 +604,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// (for instance to move the file after we have processed it)
GenericFileOnCompletion<T> onCompletion
= new GenericFileOnCompletion<>(endpoint, operations,
processStrategy, target, absoluteFileName);
- onCompletion.setKeptDoneFiles(getKeptDoneFiles());
+ GenericFileKeptDoneFiles kept = getKeptDoneFiles();
+ if (kept != null) {
+ onCompletion.setKeptDoneFiles(kept.batch());
+ }
exchange.getExchangeExtension().addOnCompletion(onCompletion);
LOG.debug("About to process file: {} using exchange: {}", target,
exchange);
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java
index e9998adddedb..d3820718a525 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java
@@ -31,18 +31,25 @@ import java.util.Set;
* file that was taken and failed is not left: as before, the done file is
deleted after the last file of the poll
* unless that file failed.
* <p/>
- * The state is in memory and cleared when the consumer stops.
+ * The files of a poll can complete on another thread while the next poll runs
(the route continues asynchronously), so
+ * the last file of a poll checks the done files kept by its own poll as well
as those kept by the current poll (see
+ * {@link Batch}).
+ * <p/>
+ * The state is in memory, and the done files kept by the current poll are
forgotten when the consumer stops.
*/
final class GenericFileKeptDoneFiles {
- private final Set<String> kept = new HashSet<>();
+ // the done files kept by the current poll: a complete listing starts a
new set, and the set of the previous poll
+ // stays with the files of that poll that have not completed yet
+ private Set<String> kept = new HashSet<>();
/**
* Called once the directory is listed, before the files are consumed.
When the listing was complete, forgets the
- * done files kept by the last poll, as the files of this poll decide
again whether they are kept. When the listing
- * was cut (it stopped at the limit), the kept done files are left as they
are: a done file not listed is still
- * kept, as its files may be among those not listed, and the caller then
keeps the done file of every listed file,
- * as the poll leaves files for the next one.
+ * done files kept by the last poll, as the files of this poll decide
again whether they are kept; the files of the
+ * last poll that have not completed yet still find them kept (see {@link
#batch()}). When the listing was cut (it
+ * stopped at the limit), the kept done files are left as they are: a done
file not listed is still kept, as its
+ * files may be among those not listed, and the caller then keeps the done
file of every listed file, as the poll
+ * leaves files for the next one.
*
* @param listedDoneFiles the done files of the files this poll found
* @param listingComplete whether every file was listed
@@ -60,7 +67,7 @@ final class GenericFileKeptDoneFiles {
answer.add(doneFileName);
}
}
- kept.clear();
+ kept = new HashSet<>();
}
return answer;
}
@@ -73,13 +80,43 @@ final class GenericFileKeptDoneFiles {
}
/**
- * Whether the done file must be kept after the last file of the poll.
+ * Whether the current poll keeps the done file (the files of a poll check
it through {@link #batch()}).
*/
synchronized boolean isKept(String doneFileName) {
return kept.contains(doneFileName);
}
+ /**
+ * The kept done files as the files of the current poll see them, to take
when a file of this poll is started.
+ */
+ synchronized Batch batch() {
+ return new Batch(kept);
+ }
+
synchronized void clear() {
kept.clear();
}
+
+ /**
+ * The kept done files as the files of one poll see them.
+ */
+ final class Batch {
+
+ private final Set<String> keptByBatch;
+
+ private Batch(Set<String> keptByBatch) {
+ this.keptByBatch = keptByBatch;
+ }
+
+ /**
+ * Whether the done file must be kept after the last file of this
poll: this poll or the current one left files
+ * of its directory. A file of this poll that completes while the next
poll runs must not delete a done file
+ * that this poll kept before the next poll has decided again whether
it keeps it.
+ */
+ boolean isKept(String doneFileName) {
+ synchronized (GenericFileKeptDoneFiles.this) {
+ return keptByBatch.contains(doneFileName) ||
kept.contains(doneFileName);
+ }
+ }
+ }
}
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java
index bd5529765ac2..82f3cdfd1fd9 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java
@@ -40,7 +40,7 @@ public class GenericFileOnCompletion<T> implements
Synchronization {
private ExceptionHandler exceptionHandler;
private final GenericFile<T> file;
private final String absoluteFileName;
- private GenericFileKeptDoneFiles keptDoneFiles;
+ private GenericFileKeptDoneFiles.Batch keptDoneFiles;
public GenericFileOnCompletion(GenericFileEndpoint<T> endpoint,
GenericFileOperations<T> operations,
GenericFileProcessStrategy processStrategy,
GenericFile<T> file,
@@ -67,10 +67,10 @@ public class GenericFileOnCompletion<T> implements
Synchronization {
}
/**
- * The done files kept for the files the poll left, when the endpoint uses
a done file with a fixed name;
- * <tt>null</tt> otherwise.
+ * The done files kept for the files the poll left, as the files of the
poll of this file see them, when the
+ * endpoint uses a done file with a fixed name; <tt>null</tt> otherwise.
*/
- void setKeptDoneFiles(GenericFileKeptDoneFiles keptDoneFiles) {
+ void setKeptDoneFiles(GenericFileKeptDoneFiles.Batch keptDoneFiles) {
this.keptDoneFiles = keptDoneFiles;
}
diff --git
a/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileKeptDoneFilesTest.java
b/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileKeptDoneFilesTest.java
index 37960c00481c..750cc4494dfe 100644
---
a/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileKeptDoneFilesTest.java
+++
b/components/camel-file/src/test/java/org/apache/camel/component/file/GenericFileKeptDoneFilesTest.java
@@ -55,4 +55,26 @@ class GenericFileKeptDoneFilesTest {
assertEquals(List.of("dir/ready"), keptDoneFiles.startBatch(Set.of(),
true));
assertFalse(keptDoneFiles.isKept("dir/ready"));
}
+
+ @Test
+ void fileOfTheLastPollFindsItsKeptDoneFileWhileTheNextPollDecidesAgain() {
+ GenericFileKeptDoneFiles.Batch lastPoll = keptDoneFiles.batch();
+ keptDoneFiles.keep("dir/ready");
+ assertTrue(lastPoll.isKept("dir/ready"));
+
+ // the next poll lists the directory completely: until it keeps the
done file again for the files it leaves, a
+ // file of the last poll that completes on another thread must still
find the done file kept by its poll
+ assertNull(keptDoneFiles.startBatch(Set.of("dir/ready"), true));
+ assertFalse(keptDoneFiles.isKept("dir/ready"));
+ assertTrue(lastPoll.isKept("dir/ready"));
+
+ // the files of the next poll decide again: it leaves no file, so its
last file deletes the done file
+ GenericFileKeptDoneFiles.Batch nextPoll = keptDoneFiles.batch();
+ assertFalse(nextPoll.isKept("dir/ready"));
+
+ // a done file kept by the next poll is kept for the files of the last
poll as well
+ keptDoneFiles.keep("other/ready");
+ assertTrue(nextPoll.isKept("other/ready"));
+ assertTrue(lastPoll.isKept("other/ready"));
+ }
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameAsyncLastFileTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameAsyncLastFileTest.java
new file mode 100644
index 000000000000..91fa9f91dccc
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameAsyncLastFileTest.java
@@ -0,0 +1,134 @@
+/*
+ * 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.File;
+import java.nio.file.Files;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.support.DefaultPollingConsumerPollStrategy;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * The last file of a poll that left files completes on another thread while
the next poll runs. The next poll lists the
+ * directory completely and leaves a file too, but it keeps the done file
again only after it sorted its files. The last
+ * file of the first poll completes in between, and must not delete the done
file, or the file left by the next poll
+ * waits for a new done file.
+ */
+public class FileConsumerDoneFileNameAsyncLastFileTest extends
ContextTestSupport {
+
+ private final AtomicInteger polls = new AtomicInteger();
+ private final AtomicBoolean secondPollSorting = new AtomicBoolean();
+ private final CountDownLatch bStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseB = new CountDownLatch(1);
+ private volatile NotifyBuilder bDone;
+ private volatile Boolean doneFileAfterB;
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry registry = super.createCamelRegistry();
+ registry.bind("countPolls", new DefaultPollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ polls.incrementAndGet();
+ return true;
+ }
+ });
+ // sorts by name; the sorter is called after the listing and before
the done file is kept for the files cut by
+ // maxMessagesPerPoll, so in the second poll it lets b.txt, the last
file of the first poll, complete there
+ Comparator<GenericFile<File>> sorter = (f1, f2) -> {
+ if (polls.get() == 2 && secondPollSorting.compareAndSet(false,
true)) {
+ letBComplete();
+ }
+ return f1.getFileName().compareTo(f2.getFileName());
+ };
+ registry.bind("byNameLettingBComplete", sorter);
+ return registry;
+ }
+
+ private void letBComplete() {
+ try {
+ if (bStarted.await(20, TimeUnit.SECONDS)) {
+ releaseB.countDown();
+ if (bDone.matches(20, TimeUnit.SECONDS)) {
+ doneFileAfterB = Files.exists(testFile("async/ready"));
+ }
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Test
+ public void
testLastFileOfPollCompletingWhileNextPollDecidesKeepsDoneFile() throws
Exception {
+ Files.createDirectories(testDirectory("async"));
+ for (String name : List.of("a", "b", "c", "d", "e")) {
+ Files.writeString(testFile("async/" + name + ".txt"),
name.toUpperCase());
+ }
+ Files.writeString(testFile("async/ready"), "");
+ bDone = new
NotifyBuilder(context).whenAnyDoneMatches(header(Exchange.FILE_NAME_ONLY).isEqualTo("b.txt")).create();
+ getMockEndpoint("mock:async").expectedBodiesReceivedInAnyOrder("A",
"B", "C", "D", "E");
+
+ // the first poll takes a.txt and b.txt and leaves c.txt, d.txt and
e.txt; the second takes c.txt and d.txt and
+ // leaves e.txt, so the done file must stay until the third poll
+ context.getRouteController().startRoute("async");
+
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() ->
doneFileAfterB != null);
+ assertEquals(Boolean.TRUE, doneFileAfterB,
+ "The done file must not be deleted by b.txt, as the second
poll leaves e.txt");
+ assertMockEndpointsSatisfied();
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(
+ () -> assertFalse(Files.exists(testFile("async/ready")), "The
done file must be deleted after the last file"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("async?doneFileName=ready&maxMessagesPerPoll=2&eagerMaxMessagesPerPoll=false"
+ +
"&sorter=#byNameLettingBComplete&pollStrategy=#countPolls&initialDelay=0&delay=10"))
+ .routeId("async").autoStartup(false)
+ .threads()
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("B".equals(e.getMessage().getBody())) {
+ bStarted.countDown();
+ releaseB.await(20, TimeUnit.SECONDS);
+ }
+ })
+ .to("mock:async");
+ }
+ };
+ }
+}
