From f959c794db82c189eab63c3419eb3a5859136d02 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Sat, 17 Oct 2026 14:17:46 +0530 Subject: [PATCH 1/2] CAMEL-25728: camel-file - keep a done file with a fixed name for the files a poll left With doneFileName=ready the consumer only picks up the files of a directory while the done file exists, and the last file of the batch (CamelBatchComplete) deletes it. The batch is the poll: with maxMessagesPerPoll the files beyond the limit, and a file whose processing could not start (read lock not acquired), stay in the directory, but the done file is gone, so they are left until a new done file is written. The poll now remembers the files it left (beyond the limit, or not started), per done file, and the last file of the batch does not delete the done file while there are any. The next poll only consumes these files, so a file written to the directory meanwhile still waits for a new done file, and the done file is deleted after its last file as before (or by the next poll if these files are gone). A file that was processed and failed is not left: it is handled as before. To know the files beyond the limit, such a consumer checks every file of the directory also with the eager limit, and then cuts the list to the same files the eager limit takes. The state is in memory and cleared on stop. Dynamic done files, noop and recursive=true are unchanged. The FTP, SFTP, SMB and Azure Files consumers share this code. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/docs/file-component.adoc | 4 + .../src/main/docs/file-component.adoc | 4 + .../component/file/GenericFileConsumer.java | 86 ++++- .../file/GenericFileDoneFileAdmission.java | 88 +++++ .../file/GenericFileOnCompletion.java | 14 + ...FileConsumerDoneFileNameFilesLeftTest.java | 356 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 8 + 7 files changed, 556 insertions(+), 4 deletions(-) create mode 100644 components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java create mode 100644 core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameFilesLeftTest.java 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 8e7e55ca5ec58..21d664f02d761 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,10 @@ 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 a file +written to the directory after the _done file_ waits for a new _done file_. [NOTE] ==== diff --git a/components/camel-file/src/main/docs/file-component.adoc b/components/camel-file/src/main/docs/file-component.adoc index 8e7e55ca5ec58..21d664f02d761 100644 --- a/components/camel-file/src/main/docs/file-component.adoc +++ b/components/camel-file/src/main/docs/file-component.adoc @@ -765,6 +765,10 @@ 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 a file +written to the directory after the _done file_ waits for a new _done file_. [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 493ad60672342..99ae21bbe3749 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.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 extends ScheduledBatchPollingConsum private boolean retrieveFile = true; // files already reported as resolving outside the starting directory (bounded) private final Map skippedOutsideStartingDirectory = LRUCacheFactory.newLRUCache(1000); + // the files a poll left while a done file with a fixed name was there (null if the endpoint has no such done file) + private final GenericFileDoneFileAdmission doneFileAdmission; protected GenericFileConsumer(GenericFileEndpoint endpoint, Processor processor, GenericFileOperations operations, GenericFileProcessStrategy processStrategy) { @@ -79,6 +83,10 @@ protected GenericFileConsumer(GenericFileEndpoint endpoint, Processor process 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.doneFileAdmission = endpoint.getDoneFileName() != null && !endpoint.isNoop() + && !endpoint.getDoneFileName().contains("{file:name") ? new GenericFileDoneFileAdmission() : null; } public Processor getCustomProcessor() { @@ -165,6 +173,25 @@ protected int poll(Exchange dynamic) throws Exception { LOG.debug("Took {} to poll: {}", TimeUtils.printDuration(delta, true), name); } + // forget the files left by the last poll, they are listed again if still there + GenericFileDoneFileAdmission admission = getDoneFileAdmission(); + if (admission != null) { + startDoneFileBatch(admission, files); + } + + // the files beyond the limit are listed when a done file with a fixed name is used (see canPollMoreFiles), so + // cut the list as the eager limit does (same files) and leave the others for the next poll + if (admission != null && eagerLimitMaxMessagesPerPoll && maxMessagesPerPoll > 0 + && files.size() > maxMessagesPerPoll) { + List> excessive = files.subList(maxMessagesPerPoll, files.size()); + removeExcessiveInProgressFiles(excessive); + for (GenericFile file : excessive) { + doneFileLeft(file); + } + excessive.clear(); + limitHit = true; + } + // 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.", @@ -259,6 +286,8 @@ public int processBatch(Queue exchanges) { // 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 + doneFileLeft(getExchangeFileProperty(exchange)); answer--; // this exchange was not started processing so remember to release it afterward notStarted.add(exchange); @@ -287,6 +316,7 @@ protected void removeExcessiveInProgressFiles(Deque exchanges, int lim GenericFile file = exchange.getProperty(ExchangePropertyKey.FILE_EXCHANGE_FILE, GenericFile.class); String key = file.getAbsoluteFilePath(); endpoint.getInProgressRepository().remove(key); + doneFileLeft(file); // if we added eager to idempotent then we need to remove this if (Boolean.TRUE.equals(endpoint.isIdempotent()) && endpoint.isIdempotentEager() && endpoint.getIdempotentRepository() != null) { @@ -323,6 +353,41 @@ protected void removeExcessiveInProgressFiles(List> files) { } } + /** + * The files left by the last poll while a done file with a fixed name was there, or null if not used (no + * such done file, or a custom processor that does not complete the files). + */ + GenericFileDoneFileAdmission getDoneFileAdmission() { + return customProcessor == null ? doneFileAdmission : null; + } + + private void startDoneFileBatch(GenericFileDoneFileAdmission admission, List> files) { + Set doneFiles = new HashSet<>(); + for (GenericFile file : files) { + doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath())); + } + List gone = admission.startBatch(doneFiles); + 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); + } + } + } + } + + private void doneFileLeft(GenericFile file) { + GenericFileDoneFileAdmission admission = getDoneFileAdmission(); + if (admission != null) { + String absoluteFileName = file.getAbsoluteFilePath(); + admission.left(endpoint.createDoneFileName(absoluteFileName), absoluteFileName); + } + } + /** * Whether or not we can continue polling for more files * @@ -330,8 +395,9 @@ protected void removeExcessiveInProgressFiles(List> files) { * @return true to continue, false to stop due hitting maxMessagesPerPoll limit */ public boolean canPollMoreFiles(List fileList) { - // at this point we should not limit if we are not eager - if (!eagerLimitMaxMessagesPerPoll) { + // at this point we should not limit if we are not eager, or when a done file with a fixed name is used, as + // the files beyond the limit must be known to keep the done file for them (the list is cut after polling) + if (!eagerLimitMaxMessagesPerPoll || getDoneFileAdmission() != null) { return true; } @@ -483,8 +549,10 @@ protected boolean processExchange(final Exchange exchange) { // 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 onCompletion + = new GenericFileOnCompletion<>(endpoint, operations, processStrategy, target, absoluteFileName); + onCompletion.setDoneFileAdmission(getDoneFileAdmission()); + exchange.getExchangeExtension().addOnCompletion(onCompletion); LOG.debug("About to process file: {} using exchange: {}", target, exchange); @@ -914,6 +982,13 @@ protected boolean isMatched( if (!isMatched(file, doneFileName, files)) { return false; } + + // a done file kept for files left by the last poll only admits these files, not the ones written after it + GenericFileDoneFileAdmission admission = getDoneFileAdmission(); + if (admission != null && !admission.isAdmitted(doneFileName, absoluteFilePath)) { + LOG.trace("Skipping file: {} as it is not admitted by the done file: {}", name, doneFileName); + return false; + } } return true; @@ -1011,6 +1086,9 @@ protected void doStart() throws Exception { protected void doStop() throws Exception { prepareOnStartup = false; skippedOutsideStartingDirectory.clear(); + if (doneFileAdmission != null) { + doneFileAdmission.clear(); + } super.doStop(); ServiceHelper.stopService(processStrategy); } diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java new file mode 100644 index 0000000000000..fe599bf8fd2d7 --- /dev/null +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java @@ -0,0 +1,88 @@ +/* + * 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.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * The files that a poll left behind while a done file with a fixed name (such as {@code doneFileName=ready}) was there, + * per done file (that is per directory), for one consumer. + *

+ * 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 done file is then kept for the next poll, which only consumes the + * files left: a file written after the done file waits for a new done file, as on a done file that had been deleted. A + * file that was taken and failed is not left: as before, it waits for the next done file. + *

+ * The state is in memory and cleared when the consumer stops. + */ +final class GenericFileDoneFileAdmission { + + // done file -> files left by the last poll + private final Map> left = new HashMap<>(); + + /** + * Whether the file may be consumed: all files may, unless the last poll left files of this done file, then only + * these. + */ + synchronized boolean isAdmitted(String doneFileName, String absoluteFileName) { + Set files = left.get(doneFileName); + return files == null || files.contains(absoluteFileName); + } + + /** + * Called once the directory is listed, before the files are consumed: forgets the files left by the last poll. + * + * @param listedDoneFiles the done files of the files this poll found + * @return the done files kept for files that are gone (not found again), to delete + */ + synchronized List startBatch(Set listedDoneFiles) { + List answer = null; + for (String doneFileName : left.keySet()) { + if (!listedDoneFiles.contains(doneFileName)) { + if (answer == null) { + answer = new ArrayList<>(); + } + answer.add(doneFileName); + } + } + left.clear(); + return answer; + } + + /** + * Called for a file the poll did not take. + */ + synchronized void left(String doneFileName, String absoluteFileName) { + left.computeIfAbsent(doneFileName, k -> new HashSet<>()).add(absoluteFileName); + } + + /** + * Whether the poll left files of this done file, which must then be kept. + */ + synchronized boolean hasLeft(String doneFileName) { + return left.containsKey(doneFileName); + } + + synchronized void clear() { + left.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 9f6cfbad39e2b..4b02686c1df4f 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 implements Synchronization { private ExceptionHandler exceptionHandler; private final GenericFile file; private final String absoluteFileName; + private GenericFileDoneFileAdmission doneFileAdmission; public GenericFileOnCompletion(GenericFileEndpoint endpoint, GenericFileOperations operations, GenericFileProcessStrategy processStrategy, GenericFile file, @@ -65,6 +66,13 @@ public void onFailure(Exchange exchange) { onCompletion(exchange); } + /** + * The files left by the poll, when the endpoint uses a done file with a fixed name; null otherwise. + */ + void setDoneFileAdmission(GenericFileDoneFileAdmission doneFileAdmission) { + this.doneFileAdmission = doneFileAdmission; + } + public ExceptionHandler getExceptionHandler() { return exceptionHandler; } @@ -185,6 +193,12 @@ protected void handleDoneFile(Exchange exchange) { // 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 && doneFileAdmission != null && doneFileAdmission.hasLeft(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 0000000000000..7df330d302093 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerDoneFileNameFilesLeftTest.java @@ -0,0 +1,356 @@ +/* + * 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.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +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; + +/** + * A done file with a fixed name stands for the files of its directory that were there when it was written. It must not + * be deleted while the poll left some of these files for the next one (beyond maxMessagesPerPoll, or not started), or + * they wait for a new done file; and a file written after it must still wait for a new done file. A file that fails is + * handled as before: it waits for the next done file. + *

+ * 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); + + @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 operations, GenericFile file, Exchange exchange) + throws Exception { + if ("b.txt".equals(file.getFileNameOnly()) && lockB.getAndSet(false)) { + return false; + } + return super.acquireExclusiveReadLock(operations, file, exchange); + } + }); + return registry; + } + + @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 testFileWrittenWhileFilesAreLeftWaitsForNewDoneFile() throws Exception { + // x.txt is written while a.txt is routed: b.txt and c.txt are left for later polls, x.txt is not admitted + writeFilesAndDoneFile("gate"); + MockEndpoint mock = getMockEndpoint("mock:gate"); + mock.expectedBodiesReceivedInAnyOrder("A", "B", "C"); + + context.getRouteController().startRoute("gate"); + + awaitDoneFileDeleted("gate/ready"); + mock.assertIsSatisfied(); + assertFileExists(testFile("gate/x.txt")); + + // a new done file admits x.txt + mock.reset(); + mock.expectedBodiesReceived("X"); + template.sendBodyAndHeader(fileUri("gate"), "", Exchange.FILE_NAME, "ready"); + + mock.assertIsSatisfied(); + awaitDoneFileDeleted("gate/ready"); + } + + @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 + writeFilesAndDoneFile("poisonlimit", "BAD"); + MockEndpoint mock = getMockEndpoint("mock:poisonlimit"); + mock.expectedBodiesReceivedInAnyOrder("A", "C"); + + context.getRouteController().startRoute("poisonlimit"); + + awaitDoneFileDeleted("poisonlimit/ready"); + mock.assertIsSatisfied(); + + 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("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())) { + 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 f1cc998b316ea..f82725b371fec 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,14 @@ 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 only these files are consumed. With `maxMessagesPerPoll` and `eagerMaxMessagesPerPoll=true` (the +default), such a consumer checks every file of the directory on each poll, as `eagerMaxMessagesPerPoll=false` does. +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 --output=

` now writes one normalized file per input file into the From e47aeb82b34e5939b735df666e78daf8bb574f05 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Sun, 18 Oct 2026 15:12:20 +0530 Subject: [PATCH 2/2] CAMEL-25728: camel-file - keep the done file without gating new files, list one file past the limit Review follow-up: - the kept done file no longer admits only the files left: the polls consume every file of the directory while it is there, as they do whenever a done file is there. A Camel producer with doneFileName rewrites the done file after each file; with the gate its files written during the drain were rejected and then stranded when the done file was deleted. A failed last file is now tried again with the files left, as on main. - with the eager limit, one file past maxMessagesPerPoll is listed (instead of every file) to know that the poll leaves files; when the listing stops there, the done files of the files listed are kept, and a kept done file is only deleted as gone after a complete listing. - the done file kept for a file that was not started is the one of the file as polled, not of the file the preMove bound to the exchange. - upgrade guide: pollEnrich and ConsumerTemplate.receive keep the done file until the last file. Co-Authored-By: Claude Opus 5.5 --- .../camel/catalog/docs/file-component.adoc | 5 +- .../src/main/docs/file-component.adoc | 5 +- .../component/file/GenericFileConsumer.java | 104 +++++++------ .../file/GenericFileDoneFileAdmission.java | 88 ----------- .../file/GenericFileKeptDoneFiles.java | 85 +++++++++++ .../file/GenericFileOnCompletion.java | 11 +- ...FileConsumerDoneFileNameFilesLeftTest.java | 144 ++++++++++++++++-- .../pages/camel-4x-upgrade-guide-4_24.adoc | 7 +- 8 files changed, 289 insertions(+), 160 deletions(-) delete mode 100644 components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java create mode 100644 components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java 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 21d664f02d761..f695acd43a259 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 @@ -767,8 +767,9 @@ 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 a file -written to the directory after the _done file_ waits for a new _done file_. +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 21d664f02d761..f695acd43a259 100644 --- a/components/camel-file/src/main/docs/file-component.adoc +++ b/components/camel-file/src/main/docs/file-component.adoc @@ -767,8 +767,9 @@ 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 a file -written to the directory after the _done file_ waits for a new _done file_. +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 99ae21bbe3749..f26f005fcffa7 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 @@ -69,8 +69,8 @@ public abstract class GenericFileConsumer extends ScheduledBatchPollingConsum private boolean retrieveFile = true; // files already reported as resolving outside the starting directory (bounded) private final Map skippedOutsideStartingDirectory = LRUCacheFactory.newLRUCache(1000); - // the files a poll left while a done file with a fixed name was there (null if the endpoint has no such done file) - private final GenericFileDoneFileAdmission doneFileAdmission; + // 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 endpoint, Processor processor, GenericFileOperations operations, GenericFileProcessStrategy processStrategy) { @@ -85,8 +85,8 @@ protected GenericFileConsumer(GenericFileEndpoint endpoint, Processor process 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.doneFileAdmission = endpoint.getDoneFileName() != null && !endpoint.isNoop() - && !endpoint.getDoneFileName().contains("{file:name") ? new GenericFileDoneFileAdmission() : null; + this.keptDoneFiles = endpoint.getDoneFileName() != null && !endpoint.isNoop() + && !endpoint.getDoneFileName().contains("{file:name") ? new GenericFileKeptDoneFiles() : null; } public Processor getCustomProcessor() { @@ -173,23 +173,20 @@ protected int poll(Exchange dynamic) throws Exception { LOG.debug("Took {} to poll: {}", TimeUtils.printDuration(delta, true), name); } - // forget the files left by the last poll, they are listed again if still there - GenericFileDoneFileAdmission admission = getDoneFileAdmission(); - if (admission != null) { - startDoneFileBatch(admission, files); - } - - // the files beyond the limit are listed when a done file with a fixed name is used (see canPollMoreFiles), so - // cut the list as the eager limit does (same files) and leave the others for the next poll - if (admission != null && eagerLimitMaxMessagesPerPoll && maxMessagesPerPoll > 0 - && files.size() > maxMessagesPerPoll) { - List> excessive = files.subList(maxMessagesPerPoll, files.size()); - removeExcessiveInProgressFiles(excessive); - for (GenericFile file : excessive) { - doneFileLeft(file); + 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 listedDoneFiles = startDoneFileBatch(kept, files, !listingCut); + if (listingCut) { + List> excessive = files.subList(maxMessagesPerPoll, files.size()); + removeExcessiveInProgressFiles(excessive); + excessive.clear(); + limitHit = true; + listedDoneFiles.forEach(kept::keep); } - excessive.clear(); - limitHit = true; } // log if we hit the limit @@ -228,6 +225,10 @@ protected int poll(Exchange dynamic) throws Exception { 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); @@ -274,6 +275,9 @@ public int processBatch(Queue exchanges) { // 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) { @@ -287,13 +291,18 @@ public int processBatch(Queue exchanges) { // 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 - doneFileLeft(getExchangeFileProperty(exchange)); + 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); @@ -316,7 +325,6 @@ protected void removeExcessiveInProgressFiles(Deque exchanges, int lim GenericFile file = exchange.getProperty(ExchangePropertyKey.FILE_EXCHANGE_FILE, GenericFile.class); String key = file.getAbsoluteFilePath(); endpoint.getInProgressRepository().remove(key); - doneFileLeft(file); // if we added eager to idempotent then we need to remove this if (Boolean.TRUE.equals(endpoint.isIdempotent()) && endpoint.isIdempotentEager() && endpoint.getIdempotentRepository() != null) { @@ -354,19 +362,25 @@ protected void removeExcessiveInProgressFiles(List> files) { } /** - * The files left by the last poll while a done file with a fixed name was there, or null if not used (no - * such done file, or a custom processor that does not complete the files). + * The done files with a fixed name kept for the files left by the last poll, or null if not used (no such + * done file, or a custom processor that does not complete the files). */ - GenericFileDoneFileAdmission getDoneFileAdmission() { - return customProcessor == null ? doneFileAdmission : null; + GenericFileKeptDoneFiles getKeptDoneFiles() { + return customProcessor == null ? keptDoneFiles : null; } - private void startDoneFileBatch(GenericFileDoneFileAdmission admission, List> files) { + /** + * 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 startDoneFileBatch(GenericFileKeptDoneFiles kept, List> files, boolean complete) { Set doneFiles = new HashSet<>(); for (GenericFile file : files) { doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath())); } - List gone = admission.startBatch(doneFiles); + List 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) { @@ -378,13 +392,13 @@ private void startDoneFileBatch(GenericFileDoneFileAdmission admission, List file) { - GenericFileDoneFileAdmission admission = getDoneFileAdmission(); - if (admission != null) { - String absoluteFileName = file.getAbsoluteFilePath(); - admission.left(endpoint.createDoneFileName(absoluteFileName), absoluteFileName); + private void keepDoneFile(GenericFile file) { + GenericFileKeptDoneFiles kept = getKeptDoneFiles(); + if (kept != null) { + kept.keep(endpoint.createDoneFileName(file.getAbsoluteFilePath())); } } @@ -395,9 +409,8 @@ private void doneFileLeft(GenericFile file) { * @return true to continue, false to stop due hitting maxMessagesPerPoll limit */ public boolean canPollMoreFiles(List fileList) { - // at this point we should not limit if we are not eager, or when a done file with a fixed name is used, as - // the files beyond the limit must be known to keep the done file for them (the list is cut after polling) - if (!eagerLimitMaxMessagesPerPoll || getDoneFileAdmission() != null) { + // at this point we should not limit if we are not eager + if (!eagerLimitMaxMessagesPerPoll) { return true; } @@ -406,6 +419,12 @@ public boolean canPollMoreFiles(List fileList) { 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; } @@ -551,7 +570,7 @@ protected boolean processExchange(final Exchange exchange) { // (for instance to move the file after we have processed it) GenericFileOnCompletion onCompletion = new GenericFileOnCompletion<>(endpoint, operations, processStrategy, target, absoluteFileName); - onCompletion.setDoneFileAdmission(getDoneFileAdmission()); + onCompletion.setKeptDoneFiles(getKeptDoneFiles()); exchange.getExchangeExtension().addOnCompletion(onCompletion); LOG.debug("About to process file: {} using exchange: {}", target, exchange); @@ -982,13 +1001,6 @@ protected boolean isMatched( if (!isMatched(file, doneFileName, files)) { return false; } - - // a done file kept for files left by the last poll only admits these files, not the ones written after it - GenericFileDoneFileAdmission admission = getDoneFileAdmission(); - if (admission != null && !admission.isAdmitted(doneFileName, absoluteFilePath)) { - LOG.trace("Skipping file: {} as it is not admitted by the done file: {}", name, doneFileName); - return false; - } } return true; @@ -1086,8 +1098,8 @@ protected void doStart() throws Exception { protected void doStop() throws Exception { prepareOnStartup = false; skippedOutsideStartingDirectory.clear(); - if (doneFileAdmission != null) { - doneFileAdmission.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/GenericFileDoneFileAdmission.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java deleted file mode 100644 index fe599bf8fd2d7..0000000000000 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileDoneFileAdmission.java +++ /dev/null @@ -1,88 +0,0 @@ -/* - * 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.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; - -/** - * The files that a poll left behind while a done file with a fixed name (such as {@code doneFileName=ready}) was there, - * per done file (that is per directory), for one consumer. - *

- * 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 done file is then kept for the next poll, which only consumes the - * files left: a file written after the done file waits for a new done file, as on a done file that had been deleted. A - * file that was taken and failed is not left: as before, it waits for the next done file. - *

- * The state is in memory and cleared when the consumer stops. - */ -final class GenericFileDoneFileAdmission { - - // done file -> files left by the last poll - private final Map> left = new HashMap<>(); - - /** - * Whether the file may be consumed: all files may, unless the last poll left files of this done file, then only - * these. - */ - synchronized boolean isAdmitted(String doneFileName, String absoluteFileName) { - Set files = left.get(doneFileName); - return files == null || files.contains(absoluteFileName); - } - - /** - * Called once the directory is listed, before the files are consumed: forgets the files left by the last poll. - * - * @param listedDoneFiles the done files of the files this poll found - * @return the done files kept for files that are gone (not found again), to delete - */ - synchronized List startBatch(Set listedDoneFiles) { - List answer = null; - for (String doneFileName : left.keySet()) { - if (!listedDoneFiles.contains(doneFileName)) { - if (answer == null) { - answer = new ArrayList<>(); - } - answer.add(doneFileName); - } - } - left.clear(); - return answer; - } - - /** - * Called for a file the poll did not take. - */ - synchronized void left(String doneFileName, String absoluteFileName) { - left.computeIfAbsent(doneFileName, k -> new HashSet<>()).add(absoluteFileName); - } - - /** - * Whether the poll left files of this done file, which must then be kept. - */ - synchronized boolean hasLeft(String doneFileName) { - return left.containsKey(doneFileName); - } - - synchronized void clear() { - left.clear(); - } -} 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 0000000000000..f7a9eaa8cf0b5 --- /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. + *

+ * 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. + *

+ * The state is in memory and cleared when the consumer stops. + */ +final class GenericFileKeptDoneFiles { + + private final Set 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 startBatch(Set listedDoneFiles, boolean listingComplete) { + List 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 4b02686c1df4f..bd5529765ac23 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 implements Synchronization { private ExceptionHandler exceptionHandler; private final GenericFile file; private final String absoluteFileName; - private GenericFileDoneFileAdmission doneFileAdmission; + private GenericFileKeptDoneFiles keptDoneFiles; public GenericFileOnCompletion(GenericFileEndpoint endpoint, GenericFileOperations operations, GenericFileProcessStrategy processStrategy, GenericFile file, @@ -67,10 +67,11 @@ public void onFailure(Exchange exchange) { } /** - * The files left by the poll, when the endpoint uses a done file with a fixed name; null otherwise. + * The done files kept for the files the poll left, when the endpoint uses a done file with a fixed name; + * null otherwise. */ - void setDoneFileAdmission(GenericFileDoneFileAdmission doneFileAdmission) { - this.doneFileAdmission = doneFileAdmission; + void setKeptDoneFiles(GenericFileKeptDoneFiles keptDoneFiles) { + this.keptDoneFiles = keptDoneFiles; } public ExceptionHandler getExceptionHandler() { @@ -195,7 +196,7 @@ protected void handleDoneFile(Exchange exchange) { 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 && doneFileAdmission != null && doneFileAdmission.hasLeft(doneFileName)) { + if (complete && keptDoneFiles != null && keptDoneFiles.isKept(doneFileName)) { LOG.trace("Done file: {} is kept for the files left by this poll", doneFileName); complete = false; } 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 index 7df330d302093..34fd26d0b3572 100644 --- 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 @@ -18,11 +18,17 @@ 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; @@ -30,11 +36,14 @@ 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 stands for the files of its directory that were there when it was written. It must not - * be deleted while the poll left some of these files for the next one (beyond maxMessagesPerPoll, or not started), or - * they wait for a new done file; and a file written after it must still wait for a new done file. A file that fails is - * handled as before: it waits for the next done file. + * 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. *

* 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. @@ -45,6 +54,8 @@ public class FileConsumerDoneFileNameFilesLeftTest extends ContextTestSupport { 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 { @@ -64,6 +75,39 @@ public boolean acquireExclusiveReadLock( 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 buildFileEndpoint(String uri, String remaining, Map parameters) + throws Exception { + FileEndpoint endpoint = new FileEndpoint(uri, this) { + @Override + protected FileConsumer newFileConsumer(Processor processor, GenericFileOperations 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"); @@ -111,25 +155,77 @@ public void testDoneFileDeletedWhenExactlyMaxMessagesPerPollFiles() throws Excep } @Test - public void testFileWrittenWhileFilesAreLeftWaitsForNewDoneFile() throws Exception { - // x.txt is written while a.txt is routed: b.txt and c.txt are left for later polls, x.txt is not admitted + 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"); + 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(); - assertFileExists(testFile("gate/x.txt")); + awaitDoneFileDeleted("producer/ready"); + assertFileNotExists(testFile("producer/x.txt")); + } - // a new done file admits x.txt - mock.reset(); - mock.expectedBodiesReceived("X"); - template.sendBodyAndHeader(fileUri("gate"), "", Exchange.FILE_NAME, "ready"); + @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 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(); - awaitDoneFileDeleted("gate/ready"); + // 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 @@ -208,7 +304,8 @@ public void testFileThatAlwaysFailsDoesNotHoldDoneFile() throws Exception { @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 + // 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"); @@ -217,6 +314,7 @@ public void testFileThatAlwaysFailsWithFilesLeftDoesNotHoldDoneFile() throws Exc awaitDoneFileDeleted("poisonlimit/ready"); mock.assertIsSatisfied(); + assertEquals(2, badAttempts.get(), "b.txt is tried again with the file left"); mock.reset(); mock.expectedBodiesReceived("D"); @@ -304,6 +402,23 @@ public void configure() { }) .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) @@ -341,6 +456,7 @@ public void configure() { .convertBodyTo(String.class) .process(e -> { if ("BAD".equals(e.getMessage().getBody())) { + badAttempts.incrementAndGet(); throw new IllegalStateException("B always fails"); } }) 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 f82725b371fec..807b2cff01c28 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 @@ -207,9 +207,10 @@ So `camel validate` can report an endpoint it did not report before. 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 only these files are consumed. With `maxMessagesPerPoll` and `eagerMaxMessagesPerPoll=true` (the -default), such a consumer checks every file of the directory on each poll, as `eagerMaxMessagesPerPoll=false` does. -This applies to camel-ftp, camel-sftp, camel-smb and camel-azure-files as well. +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