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 0d9b605765ff CAMEL-25728: camel-file - keep a done file with a fixed
name for the files a poll left (#28035)
0d9b605765ff is described below
commit 0d9b605765ff9f302648c8a07e0cd62afe38e92f
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 00:35:35 2026 +0530
CAMEL-25728: camel-file - keep a done file with a fixed name for the files
a poll left (#28035)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/file-component.adoc | 5 +
.../camel-file/src/main/docs/file-component.adoc | 5 +
.../camel/component/file/GenericFileConsumer.java | 94 +++-
.../component/file/GenericFileKeptDoneFiles.java | 85 ++++
.../component/file/GenericFileOnCompletion.java | 15 +
.../FileConsumerDoneFileNameFilesLeftTest.java | 472 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +
7 files changed, 683 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/file-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/file-component.adoc
index 8e7e55ca5ec5..f695acd43a25 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/file-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/file-component.adoc
@@ -765,6 +765,11 @@ YAML::
It will only consume files from the _bar_ folder if a _done file_ exists in
the same directory as the target files. Camel will automatically delete
the _done file_ when it's done consuming the files.
+When a poll leaves files of the directory for the next poll (beyond
+`maxMessagesPerPoll`, or not started, for example when the read lock is not
+acquired), the _done file_ is kept until they are consumed; meanwhile the
+polls consume every file of the directory, as they do whenever the _done file_
+is there.
[NOTE]
====
diff --git a/components/camel-file/src/main/docs/file-component.adoc
b/components/camel-file/src/main/docs/file-component.adoc
index 8e7e55ca5ec5..f695acd43a25 100644
--- a/components/camel-file/src/main/docs/file-component.adoc
+++ b/components/camel-file/src/main/docs/file-component.adoc
@@ -765,6 +765,11 @@ YAML::
It will only consume files from the _bar_ folder if a _done file_ exists in
the same directory as the target files. Camel will automatically delete
the _done file_ when it's done consuming the files.
+When a poll leaves files of the directory for the next poll (beyond
+`maxMessagesPerPoll`, or not started, for example when the read lock is not
+acquired), the _done file_ is kept until they are consumed; meanwhile the
+polls consume every file of the directory, as they do whenever the _done file_
+is there.
[NOTE]
====
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 493ad6067234..f26f005fcffa 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
@@ -20,10 +20,12 @@ import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
+import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.Set;
import java.util.function.Supplier;
import java.util.regex.Pattern;
@@ -67,6 +69,8 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
private boolean retrieveFile = true;
// files already reported as resolving outside the starting directory
(bounded)
private final Map<String, Boolean> skippedOutsideStartingDirectory =
LRUCacheFactory.newLRUCache(1000);
+ // the done files with a fixed name kept for the files a poll left (null
if the endpoint has no such done file)
+ private final GenericFileKeptDoneFiles keptDoneFiles;
protected GenericFileConsumer(GenericFileEndpoint<T> endpoint, Processor
processor, GenericFileOperations<T> operations,
GenericFileProcessStrategy<T>
processStrategy) {
@@ -79,6 +83,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
this.excludePattern = endpoint.getExcludePattern();
this.includeExt = endpoint.getIncludeExt() != null ?
endpoint.getIncludeExt().toLowerCase().split(",") : null;
this.excludeExt = endpoint.getExcludeExt() != null ?
endpoint.getExcludeExt().toLowerCase().split(",") : null;
+ // a done file with a fixed name stands for all the files of its
directory, and is deleted (unless noop) after
+ // the last file of a poll, so it must be kept when the poll left some
of them for the next poll
+ this.keptDoneFiles = endpoint.getDoneFileName() != null &&
!endpoint.isNoop()
+ && !endpoint.getDoneFileName().contains("{file:name") ? new
GenericFileKeptDoneFiles() : null;
}
public Processor getCustomProcessor() {
@@ -165,6 +173,22 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
LOG.debug("Took {} to poll: {}", TimeUtils.printDuration(delta,
true), name);
}
+ GenericFileKeptDoneFiles kept = getKeptDoneFiles();
+ if (kept != null) {
+ // one file more than the limit is listed when a done file with a
fixed name is used (see
+ // canPollMoreFiles): then the poll leaves files for the next one,
so cut the list to the files the eager
+ // limit takes and keep the done files of the files listed
+ boolean listingCut = eagerLimitMaxMessagesPerPoll &&
maxMessagesPerPoll > 0 && files.size() > maxMessagesPerPoll;
+ Set<String> listedDoneFiles = startDoneFileBatch(kept, files,
!listingCut);
+ if (listingCut) {
+ List<GenericFile<T>> excessive =
files.subList(maxMessagesPerPoll, files.size());
+ removeExcessiveInProgressFiles(excessive);
+ excessive.clear();
+ limitHit = true;
+ listedDoneFiles.forEach(kept::keep);
+ }
+ }
+
// log if we hit the limit
if (limitHit) {
LOG.debug("Limiting maximum messages to poll at {} files as there
were more messages in this poll.",
@@ -201,6 +225,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
if (files.size() > maxMessagesPerPoll) {
LOG.debug("Limiting maximum messages to poll at {} files as
there were more messages in this poll.",
maxMessagesPerPoll);
+ // the done files of the files cut are kept for the next poll
+ for (Exchange exchange : exchanges.subList(maxMessagesPerPoll,
exchanges.size())) {
+ keepDoneFile(getExchangeFileProperty(exchange));
+ }
// must first remove excessive files from the in progress
// repository
removeExcessiveInProgressFiles(exchanges, maxMessagesPerPoll);
@@ -247,6 +275,9 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// update pending number of exchanges
pendingExchanges = total - index - 1;
+ // the file as polled: the begin strategy may pre move it, and
then binds the pre moved file to the exchange
+ GenericFile<?> polledFile = getExchangeFileProperty(exchange);
+
// process the current exchange
boolean started;
if (customProcessor != null) {
@@ -259,12 +290,19 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// if we did not start process the file then decrement the counter
if (!started) {
+ // the done file must stay for this file, also if it is not
started as the last file of the batch
+ keepDoneFile(polledFile);
answer--;
// this exchange was not started processing so remember to
release it afterward
notStarted.add(exchange);
}
}
+ // the files not processed (the batch was stopped) are left for the
next poll too
+ for (Object exchange : exchanges) {
+ keepDoneFile(getExchangeFileProperty((Exchange) exchange));
+ }
+
// drain any in progress files as we are done with this batch
removeExcessiveInProgressFiles(CastUtils.cast((Deque<?>) notStarted,
Exchange.class), 0);
removeExcessiveInProgressFiles(CastUtils.cast((Deque<?>) exchanges,
Exchange.class), 0);
@@ -323,6 +361,47 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
}
}
+ /**
+ * The done files with a fixed name kept for the files left by the last
poll, or <tt>null</tt> if not used (no such
+ * done file, or a custom processor that does not complete the files).
+ */
+ GenericFileKeptDoneFiles getKeptDoneFiles() {
+ return customProcessor == null ? keptDoneFiles : null;
+ }
+
+ /**
+ * Starts the batch of the kept done files with the files listed, and
deletes the done files kept for files that are
+ * gone.
+ *
+ * @return the done files of the files listed
+ */
+ private Set<String> startDoneFileBatch(GenericFileKeptDoneFiles kept,
List<GenericFile<T>> files, boolean complete) {
+ Set<String> doneFiles = new HashSet<>();
+ for (GenericFile<T> file : files) {
+
doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+ }
+ List<String> gone = kept.startBatch(doneFiles, complete);
+ if (gone != null) {
+ // the done file was kept for files that are no longer there, so
it is no longer needed
+ for (String doneFileName : gone) {
+ try {
+ boolean deleted = operations.deleteFile(doneFileName);
+ LOG.trace("Done file: {} kept for files that are gone was
deleted: {}", doneFileName, deleted);
+ } catch (Exception e) {
+ LOG.debug("Cannot delete done file: {} due to: {}",
doneFileName, e.getMessage(), e);
+ }
+ }
+ }
+ return doneFiles;
+ }
+
+ private void keepDoneFile(GenericFile<?> file) {
+ GenericFileKeptDoneFiles kept = getKeptDoneFiles();
+ if (kept != null) {
+ kept.keep(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+ }
+ }
+
/**
* Whether or not we can continue polling for more files
*
@@ -340,6 +419,12 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
return true;
}
+ if (getKeptDoneFiles() != null) {
+ // with a done file with a fixed name, one more file is listed to
know whether the poll leaves files for the
+ // next one, and then the done file is kept (the list is cut to
the limit after polling)
+ return fileList.size() <= maxMessagesPerPoll;
+ }
+
// then only poll if we haven't reached the max limit
return fileList.size() < maxMessagesPerPoll;
}
@@ -483,8 +568,10 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
// register on completion callback that does the completion
// strategies
// (for instance to move the file after we have processed it)
- exchange.getExchangeExtension().addOnCompletion(
- new GenericFileOnCompletion<>(endpoint, operations,
processStrategy, target, absoluteFileName));
+ GenericFileOnCompletion<T> onCompletion
+ = new GenericFileOnCompletion<>(endpoint, operations,
processStrategy, target, absoluteFileName);
+ onCompletion.setKeptDoneFiles(getKeptDoneFiles());
+ exchange.getExchangeExtension().addOnCompletion(onCompletion);
LOG.debug("About to process file: {} using exchange: {}", target,
exchange);
@@ -1011,6 +1098,9 @@ public abstract class GenericFileConsumer<T> extends
ScheduledBatchPollingConsum
protected void doStop() throws Exception {
prepareOnStartup = false;
skippedOutsideStartingDirectory.clear();
+ if (keptDoneFiles != null) {
+ keptDoneFiles.clear();
+ }
super.doStop();
ServiceHelper.stopService(processStrategy);
}
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
new file mode 100644
index 000000000000..f7a9eaa8cf0b
--- /dev/null
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java
@@ -0,0 +1,85 @@
+/*
+ * 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.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * The done files with a fixed name (such as {@code doneFileName=ready}) that
a consumer keeps because a poll left files
+ * of their directory for the next poll.
+ * <p/>
+ * A file is left when the poll did not take it: beyond {@code
maxMessagesPerPoll}, or its processing could not start
+ * (for example the read lock was not acquired). The last file of the poll
then does not delete the done file, so the
+ * next poll consumes the files left, and any other file of the directory, as
it does whenever the done file is there. A
+ * 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.
+ */
+final class GenericFileKeptDoneFiles {
+
+ private final Set<String> kept = new HashSet<>();
+
+ /**
+ * Called once the directory is listed, before the files are consumed:
forgets the done files kept by the last poll,
+ * as the files of this poll decide again whether they are kept.
+ *
+ * @param listedDoneFiles the done files of the files this poll found
+ * @param listingComplete whether every file was listed; if not (the
listing stopped at the limit), a done file not
+ * listed is kept, as its files may be among those
not listed
+ * @return the done files kept for files that are gone
(the listing was complete and found none of
+ * their files), to delete
+ */
+ synchronized List<String> startBatch(Set<String> listedDoneFiles, boolean
listingComplete) {
+ List<String> answer = null;
+ if (listingComplete) {
+ for (String doneFileName : kept) {
+ if (!listedDoneFiles.contains(doneFileName)) {
+ if (answer == null) {
+ answer = new ArrayList<>();
+ }
+ answer.add(doneFileName);
+ }
+ }
+ kept.clear();
+ } else {
+ kept.removeAll(listedDoneFiles);
+ }
+ return answer;
+ }
+
+ /**
+ * Keeps the done file, as this poll leaves files of its directory for the
next poll.
+ */
+ synchronized void keep(String doneFileName) {
+ kept.add(doneFileName);
+ }
+
+ /**
+ * Whether the done file must be kept after the last file of the poll.
+ */
+ synchronized boolean isKept(String doneFileName) {
+ return kept.contains(doneFileName);
+ }
+
+ synchronized void clear() {
+ kept.clear();
+ }
+}
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 9f6cfbad39e2..bd5529765ac2 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,6 +40,7 @@ public class GenericFileOnCompletion<T> implements
Synchronization {
private ExceptionHandler exceptionHandler;
private final GenericFile<T> file;
private final String absoluteFileName;
+ private GenericFileKeptDoneFiles keptDoneFiles;
public GenericFileOnCompletion(GenericFileEndpoint<T> endpoint,
GenericFileOperations<T> operations,
GenericFileProcessStrategy processStrategy,
GenericFile<T> file,
@@ -65,6 +66,14 @@ public class GenericFileOnCompletion<T> implements
Synchronization {
onCompletion(exchange);
}
+ /**
+ * 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.
+ */
+ void setKeptDoneFiles(GenericFileKeptDoneFiles keptDoneFiles) {
+ this.keptDoneFiles = keptDoneFiles;
+ }
+
public ExceptionHandler getExceptionHandler() {
return exceptionHandler;
}
@@ -185,6 +194,12 @@ public class GenericFileOnCompletion<T> implements
Synchronization {
// done file must be in same path as the original input file
String doneFileName =
endpoint.createDoneFileName(absoluteFileName);
StringHelper.notEmpty(doneFileName, "doneFileName", endpoint);
+ // a done file with a fixed name is kept while the poll left files
for the next one (beyond
+ // maxMessagesPerPoll, or not started)
+ if (complete && keptDoneFiles != null &&
keptDoneFiles.isKept(doneFileName)) {
+ LOG.trace("Done file: {} is kept for the files left by this
poll", doneFileName);
+ complete = false;
+ }
// we should delete the dynamic done file
if (endpoint.getDoneFileName().contains("{file:name") || complete)
{
try {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameFilesLeftTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameFilesLeftTest.java
new file mode 100644
index 000000000000..34fd26d0b357
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameFilesLeftTest.java
@@ -0,0 +1,472 @@
+/*
+ * 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.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+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.file.strategy.MarkerFileExclusiveReadLockStrategy;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.Registry;
+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.assertNull;
+
+/**
+ * A done file with a fixed name must not be deleted while the poll left files
of its directory for the next one (beyond
+ * maxMessagesPerPoll, or not started), or they wait for a new done file.
While it is kept, the next polls consume the
+ * files of the directory as they do whenever the done file is there,
including a file written meanwhile. A file that
+ * fails is handled as before: the done file is deleted after the last file of
the poll, unless that file failed.
+ * <p/>
+ * A test that expects a file not to be consumed checks it once the done file
is gone: from then on the file cannot be
+ * consumed without a new done file, and the routes are synchronous, so any
file consumed before is already received.
+ */
+public class FileConsumerDoneFileNameFilesLeftTest extends ContextTestSupport {
+
+ private final AtomicBoolean failB = new AtomicBoolean(true);
+ private final AtomicBoolean failLastB = new AtomicBoolean(true);
+ private final AtomicBoolean writeLate = new AtomicBoolean(true);
+ private final AtomicBoolean lockB = new AtomicBoolean(true);
+ private final AtomicBoolean produceLate = new AtomicBoolean(true);
+ private final AtomicInteger badAttempts = new AtomicInteger();
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry registry = super.createCamelRegistry();
+ // the read lock of b.txt is not acquired the first time, so its
processing is not started
+ registry.bind("lockBOnce", new MarkerFileExclusiveReadLockStrategy() {
+ @Override
+ public boolean acquireExclusiveReadLock(
+ GenericFileOperations<File> operations, GenericFile<File>
file, Exchange exchange)
+ throws Exception {
+ if ("b.txt".equals(file.getFileNameOnly()) &&
lockB.getAndSet(false)) {
+ return false;
+ }
+ return super.acquireExclusiveReadLock(operations, file,
exchange);
+ }
+ });
+ return registry;
+ }
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ // a file component whose download of a.txt fails, after the preMove
of the begin strategy
+ context.addComponent("failingfile", new FileComponent() {
+ @Override
+ protected GenericFileEndpoint<File> buildFileEndpoint(String uri,
String remaining, Map<String, Object> parameters)
+ throws Exception {
+ FileEndpoint endpoint = new FileEndpoint(uri, this) {
+ @Override
+ protected FileConsumer newFileConsumer(Processor
processor, GenericFileOperations<File> operations) {
+ return super.newFileConsumer(processor, new
FileOperations(this) {
+ @Override
+ public boolean retrieveFile(String name, Exchange
exchange, long size) {
+ if (name.endsWith("a.txt")) {
+ throw new
GenericFileOperationFailedException("Cannot download " + name);
+ }
+ return super.retrieveFile(name, exchange,
size);
+ }
+ });
+ }
+ };
+ File file = new File(remaining);
+ endpoint.setFile(file);
+ GenericFileConfiguration config = new
GenericFileConfiguration();
+ config.setDirectory(file.getPath());
+ endpoint.setConfiguration(config);
+ return endpoint;
+ }
+ });
+ return context;
+ }
+
+ @Test
+ public void testFilesLeftByMaxMessagesPerPoll() throws Exception {
+ writeFilesAndDoneFile("limit");
+ getMockEndpoint("mock:limit").expectedBodiesReceivedInAnyOrder("A",
"B", "C");
+
+ context.getRouteController().startRoute("limit");
+
+ assertMockEndpointsSatisfied();
+ awaitDoneFileDeleted("limit/ready");
+ }
+
+ @Test
+ public void testFilesLeftByMaxMessagesPerPollNotEager() throws Exception {
+ writeFilesAndDoneFile("noteager");
+ getMockEndpoint("mock:noteager").expectedBodiesReceivedInAnyOrder("A",
"B", "C");
+
+ context.getRouteController().startRoute("noteager");
+
+ assertMockEndpointsSatisfied();
+ awaitDoneFileDeleted("noteager/ready");
+ }
+
+ @Test
+ public void testFileNotStartedIsConsumedByNextPoll() throws Exception {
+ writeFilesAndDoneFile("notstarted");
+
getMockEndpoint("mock:notstarted").expectedBodiesReceivedInAnyOrder("A", "B",
"C");
+
+ context.getRouteController().startRoute("notstarted");
+
+ assertMockEndpointsSatisfied();
+ awaitDoneFileDeleted("notstarted/ready");
+ }
+
+ @Test
+ public void testDoneFileDeletedWhenExactlyMaxMessagesPerPollFiles() throws
Exception {
+ template.sendBodyAndHeader(fileUri("exact"), "A", Exchange.FILE_NAME,
"a.txt");
+ template.sendBodyAndHeader(fileUri("exact"), "B", Exchange.FILE_NAME,
"b.txt");
+ template.sendBodyAndHeader(fileUri("exact"), "", Exchange.FILE_NAME,
"ready");
+ getMockEndpoint("mock:exact").expectedBodiesReceivedInAnyOrder("A",
"B");
+
+ context.getRouteController().startRoute("exact");
+
+ assertMockEndpointsSatisfied();
+ awaitDoneFileDeleted("exact/ready");
+ }
+
+ @Test
+ public void testFileWrittenWhileFilesAreLeftIsConsumed() throws Exception {
+ // x.txt is written while a.txt is routed: b.txt and c.txt are left
for later polls, and the done file kept for
+ // them is there, so x.txt is consumed as well (as when a poll finds a
file written after the done file)
+ writeFilesAndDoneFile("gate");
+ MockEndpoint mock = getMockEndpoint("mock:gate");
+ mock.expectedBodiesReceivedInAnyOrder("A", "B", "C", "X");
+
+ context.getRouteController().startRoute("gate");
+
+ mock.assertIsSatisfied();
+ awaitDoneFileDeleted("gate/ready");
+ assertFileNotExists(testFile("gate/x.txt"));
+ }
+
+ @Test
+ public void testFileWrittenByCamelProducerWhileFilesAreLeftIsConsumed()
throws Exception {
+ // a Camel producer with doneFileName writes x.txt and rewrites the
done file while a.txt is routed: x.txt must
+ // be consumed, not stranded once the done file is deleted after the
files left
+ writeFilesAndDoneFile("producer");
+ MockEndpoint mock = getMockEndpoint("mock:producer");
+ mock.expectedBodiesReceivedInAnyOrder("A", "B", "C", "X");
+
+ context.getRouteController().startRoute("producer");
+
+ mock.assertIsSatisfied();
+ awaitDoneFileDeleted("producer/ready");
+ assertFileNotExists(testFile("producer/x.txt"));
+ }
+
+ @Test
+ public void testConsumerTemplateReceivesEveryFileOfTheDoneFile() throws
Exception {
+ // the polling consumer takes one file per receive
(maxMessagesPerPoll=1): the done file is kept until the last
+ // file is received (before, the first receive deleted it and the
other files waited for a new done file)
+ writeFilesAndDoneFile("receive");
+ String uri = fileUri("receive?doneFileName=ready");
+
+ Set<String> bodies = new HashSet<>();
+ for (int i = 0; i < 3; i++) {
+ bodies.add(consumer.receiveBody(uri, 5000, String.class));
+ }
+
+ assertEquals(Set.of("A", "B", "C"), bodies);
+ awaitDoneFileDeleted("receive/ready");
+ assertNull(consumer.receiveBodyNoWait(uri, String.class));
+ }
+
+ @Test
+ public void
testDownloadFailingAfterPreMoveKeepsTheDoneFileOfTheDirectory() throws
Exception {
+ // the download of a.txt fails after it was pre moved to work/a.txt;
the done file kept for it must be the one of
+ // the directory polled, not work/ready, which belongs to whatever
writes to the work directory
+ writeFilesAndDoneFile("premove");
+ template.sendBodyAndHeader(fileUri("premove/work"), "",
Exchange.FILE_NAME, "ready");
+ MockEndpoint mock = getMockEndpoint("mock:premove");
+ mock.expectedBodiesReceivedInAnyOrder("B", "C");
+
+ context.getRouteController().startRoute("premove");
+
+ mock.assertIsSatisfied();
+ // a.txt is no longer in the directory, so the next poll deletes the
done file kept for it
+ awaitDoneFileDeleted("premove/ready");
+
+ // a later batch is consumed, so the polls after the failed download
have run
+ mock.reset();
+ mock.expectedBodiesReceived("D");
+ template.sendBodyAndHeader(fileUri("premove"), "D",
Exchange.FILE_NAME, "d.txt");
+ template.sendBodyAndHeader(fileUri("premove"), "", Exchange.FILE_NAME,
"ready");
+ mock.assertIsSatisfied();
+ awaitDoneFileDeleted("premove/ready");
+
+ assertFileExists(testFile("premove/work/ready"));
+ assertFileExists(testFile("premove/work/a.txt"));
+ }
+
+ @Test
+ public void
testFileWrittenDuringExactlyMaxMessagesPerPollFilesIsNotConsumed() throws
Exception {
+ // y.txt is written while b.txt, the last file, is routed, without a
done file
+ template.sendBodyAndHeader(fileUri("exactlate"), "A",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader(fileUri("exactlate"), "B",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader(fileUri("exactlate"), "",
Exchange.FILE_NAME, "ready");
+ MockEndpoint mock = getMockEndpoint("mock:exactlate");
+ mock.expectedBodiesReceived("A", "B");
+
+ context.getRouteController().startRoute("exactlate");
+
+ awaitDoneFileDeleted("exactlate/ready");
+ mock.assertIsSatisfied();
+ assertFileExists(testFile("exactlate/y.txt"));
+ }
+
+ @Test
+ public void testFileRolledBackWaitsForNextDoneFile() throws Exception {
+ // as before: b.txt fails once and waits for the next done file
+ writeFilesAndDoneFile("retry");
+ MockEndpoint mock = getMockEndpoint("mock:retry");
+ mock.expectedBodiesReceivedInAnyOrder("A", "C");
+
+ context.getRouteController().startRoute("retry");
+
+ awaitDoneFileDeleted("retry/ready");
+ mock.assertIsSatisfied();
+ assertFileExists(testFile("retry/b.txt"));
+
+ mock.reset();
+ mock.expectedBodiesReceived("B");
+ template.sendBodyAndHeader(fileUri("retry"), "", Exchange.FILE_NAME,
"ready");
+
+ mock.assertIsSatisfied();
+ awaitDoneFileDeleted("retry/ready");
+ }
+
+ @Test
+ public void testLastFileRolledBackKeepsDoneFileAsBefore() throws Exception
{
+ // as before: b.txt, the last file of the batch, fails once and writes
y.txt; the done file stays, so the next
+ // poll consumes both
+ template.sendBodyAndHeader(fileUri("rollback"), "A",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader(fileUri("rollback"), "B",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader(fileUri("rollback"), "",
Exchange.FILE_NAME, "ready");
+ getMockEndpoint("mock:rollback").expectedBodiesReceived("A", "B", "Y");
+
+ context.getRouteController().startRoute("rollback");
+
+ assertMockEndpointsSatisfied();
+ awaitDoneFileDeleted("rollback/ready");
+ }
+
+ @Test
+ public void testFileThatAlwaysFailsDoesNotHoldDoneFile() throws Exception {
+ writeFilesAndDoneFile("poison", "BAD");
+ MockEndpoint mock = getMockEndpoint("mock:poison");
+ mock.expectedBodiesReceivedInAnyOrder("A", "C");
+
+ context.getRouteController().startRoute("poison");
+
+ awaitDoneFileDeleted("poison/ready");
+ mock.assertIsSatisfied();
+
+ // the next batch is consumed with its done file (b.txt is tried again
and fails again)
+ mock.reset();
+ mock.expectedBodiesReceived("D");
+ template.sendBodyAndHeader(fileUri("poison"), "D", Exchange.FILE_NAME,
"d.txt");
+ template.sendBodyAndHeader(fileUri("poison"), "", Exchange.FILE_NAME,
"ready");
+
+ mock.assertIsSatisfied();
+ awaitDoneFileDeleted("poison/ready");
+ assertFileExists(testFile("poison/b.txt"));
+ }
+
+ @Test
+ public void testFileThatAlwaysFailsWithFilesLeftDoesNotHoldDoneFile()
throws Exception {
+ // b.txt fails as the last file of the first poll, c.txt is left for
the next one, which tries b.txt again (as
+ // the next poll does on main when the last file failed) and deletes
the done file after c.txt
+ writeFilesAndDoneFile("poisonlimit", "BAD");
+ MockEndpoint mock = getMockEndpoint("mock:poisonlimit");
+ mock.expectedBodiesReceivedInAnyOrder("A", "C");
+
+ context.getRouteController().startRoute("poisonlimit");
+
+ awaitDoneFileDeleted("poisonlimit/ready");
+ mock.assertIsSatisfied();
+ assertEquals(2, badAttempts.get(), "b.txt is tried again with the file
left");
+
+ mock.reset();
+ mock.expectedBodiesReceived("D");
+ template.sendBodyAndHeader(fileUri("poisonlimit"), "D",
Exchange.FILE_NAME, "d.txt");
+ template.sendBodyAndHeader(fileUri("poisonlimit"), "",
Exchange.FILE_NAME, "ready");
+
+ mock.assertIsSatisfied();
+ assertFileExists(testFile("poisonlimit/b.txt"));
+ }
+
+ @Test
+ public void testDynamicDoneFileNameUnchanged() throws Exception {
+ template.sendBodyAndHeader(fileUri("dynamic"), "A",
Exchange.FILE_NAME, "a.txt");
+ template.sendBodyAndHeader(fileUri("dynamic"), "", Exchange.FILE_NAME,
"a.txt.done");
+ template.sendBodyAndHeader(fileUri("dynamic"), "B",
Exchange.FILE_NAME, "b.txt");
+ template.sendBodyAndHeader(fileUri("dynamic"), "", Exchange.FILE_NAME,
"b.txt.done");
+ template.sendBodyAndHeader(fileUri("dynamic"), "C",
Exchange.FILE_NAME, "c.txt");
+ MockEndpoint mock = getMockEndpoint("mock:dynamic");
+ // the eager limit takes the first listed file, so the order is not
known
+ mock.expectedBodiesReceivedInAnyOrder("A", "B");
+
+ context.getRouteController().startRoute("dynamic");
+
+ awaitDoneFileDeleted("dynamic/a.txt.done");
+ awaitDoneFileDeleted("dynamic/b.txt.done");
+ mock.assertIsSatisfied();
+ assertFileExists(testFile("dynamic/c.txt"));
+ }
+
+ private void awaitDoneFileDeleted(String doneFile) {
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
assertFileNotExists(testFile(doneFile)));
+ }
+
+ private void writeFilesAndDoneFile(String dir) {
+ writeFilesAndDoneFile(dir, "B");
+ }
+
+ private void writeFilesAndDoneFile(String dir, String bodyB) {
+ template.sendBodyAndHeader(fileUri(dir), "A", Exchange.FILE_NAME,
"a.txt");
+ template.sendBodyAndHeader(fileUri(dir), bodyB, Exchange.FILE_NAME,
"b.txt");
+ template.sendBodyAndHeader(fileUri(dir), "C", Exchange.FILE_NAME,
"c.txt");
+ template.sendBodyAndHeader(fileUri(dir), "", Exchange.FILE_NAME,
"ready");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("limit?doneFileName=ready&maxMessagesPerPoll=2&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("limit").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:limit");
+
+
from(fileUri("noteager?doneFileName=ready&maxMessagesPerPoll=2&eagerMaxMessagesPerPoll=false"
+ + "&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("noteager").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:noteager");
+
+
from(fileUri("notstarted?doneFileName=ready&exclusiveReadLockStrategy=#lockBOnce&sortBy=file:name"
+ + "&initialDelay=0&delay=10"))
+ .routeId("notstarted").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:notstarted");
+
+
from(fileUri("retry?doneFileName=ready&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("retry").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("B".equals(e.getMessage().getBody()) &&
failB.getAndSet(false)) {
+ throw new IllegalStateException("B fails
once");
+ }
+ })
+ .to("mock:retry");
+
+
from(fileUri("exact?doneFileName=ready&maxMessagesPerPoll=2&initialDelay=0&delay=10"))
+ .routeId("exact").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:exact");
+
+
from(fileUri("gate?doneFileName=ready&maxMessagesPerPoll=1&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("gate").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if (writeLate.compareAndSet(true, false)) {
+ Files.writeString(testFile("gate/x.txt"), "X");
+ }
+ })
+ .to("mock:gate");
+
+
from(fileUri("producer?doneFileName=ready&maxMessagesPerPoll=1&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("producer").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if (produceLate.compareAndSet(true, false)) {
+ // deletes and rewrites the done file after
writing x.txt
+
template.sendBodyAndHeader(fileUri("producer?doneFileName=ready"), "X",
Exchange.FILE_NAME,
+ "x.txt");
+ }
+ })
+ .to("mock:producer");
+
+ from("failingfile:" + testDirectory("premove")
+ +
"?doneFileName=ready&preMove=work&sortBy=file:name&initialDelay=0&delay=10")
+ .routeId("premove").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:premove");
+
+
from(fileUri("exactlate?doneFileName=ready&maxMessagesPerPoll=2&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("exactlate").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("B".equals(e.getMessage().getBody())) {
+ Files.writeString(testFile("exactlate/y.txt"),
"Y");
+ }
+ })
+ .to("mock:exactlate");
+
+
from(fileUri("rollback?doneFileName=ready&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("rollback").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("B".equals(e.getMessage().getBody()) &&
failLastB.getAndSet(false)) {
+ Files.writeString(testFile("rollback/y.txt"),
"Y");
+ throw new IllegalStateException("B fails
once");
+ }
+ })
+ .to("mock:rollback");
+
+
from(fileUri("poison?doneFileName=ready&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("poison").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("BAD".equals(e.getMessage().getBody())) {
+ throw new IllegalStateException("B always
fails");
+ }
+ })
+ .to("mock:poison");
+
+
from(fileUri("poisonlimit?doneFileName=ready&maxMessagesPerPoll=2&eagerMaxMessagesPerPoll=false"
+ + "&sortBy=file:name&initialDelay=0&delay=10"))
+ .routeId("poisonlimit").autoStartup(false)
+ .convertBodyTo(String.class)
+ .process(e -> {
+ if ("BAD".equals(e.getMessage().getBody())) {
+ badAttempts.incrementAndGet();
+ throw new IllegalStateException("B always
fails");
+ }
+ })
+ .to("mock:poisonlimit");
+
+
from(fileUri("dynamic?doneFileName=${file:name}.done&maxMessagesPerPoll=1&sortBy=file:name"
+ + "&initialDelay=0&delay=10"))
+ .routeId("dynamic").autoStartup(false)
+ .convertBodyTo(String.class).to("mock:dynamic");
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 2c4a315cdf28..68a45bb42317 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -203,6 +203,15 @@ that a route sends to and no route consumes, so they
report the same endpoints o
So `camel validate` can report an endpoint it did not report before.
+=== camel-file - a done file with a fixed name is kept for the files a poll
left
+
+A done file with a fixed name (such as `doneFileName=ready`) is no longer
deleted while a poll left files of its
+directory for the next poll (beyond `maxMessagesPerPoll`, or not started);
before, they were left until a new done file
+was written. Meanwhile the polls consume every file of the directory, as they
do whenever the done file is there. This
+also applies to `pollEnrich` and `ConsumerTemplate.receive`, which take one
file per call: the done file is now kept
+until the last file of the directory is received, where before the first
received file deleted it. This applies to
+camel-ftp, camel-sftp, camel-smb and camel-azure-files as well.
+
=== camel-jbang - camel validate normalize writes one file per input file to
an output directory
`camel validate normalize <files> --output=<dir>` now writes one normalized
file per input file into the
