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..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 @@ -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 8e7e55ca5ec58..f695acd43a259 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 493ad60672342..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 @@ -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 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) { @@ -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.keptDoneFiles = endpoint.getDoneFileName() != null && !endpoint.isNoop() + && !endpoint.getDoneFileName().contains("{file:name") ? new GenericFileKeptDoneFiles() : null; } public Processor getCustomProcessor() { @@ -165,6 +173,22 @@ protected int poll(Exchange dynamic) throws Exception { 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 listedDoneFiles = startDoneFileBatch(kept, files, !listingCut); + if (listingCut) { + List> 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 @@ 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); @@ -247,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) { @@ -259,12 +290,19 @@ 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 + 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 @@ protected void removeExcessiveInProgressFiles(List> 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). + */ + 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 startDoneFileBatch(GenericFileKeptDoneFiles kept, List> files, boolean complete) { + Set doneFiles = new HashSet<>(); + for (GenericFile file : files) { + doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath())); + } + 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) { + 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 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; } @@ -483,8 +568,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.setKeptDoneFiles(getKeptDoneFiles()); + exchange.getExchangeExtension().addOnCompletion(onCompletion); LOG.debug("About to process file: {} using exchange: {}", target, exchange); @@ -1011,6 +1098,9 @@ protected void doStart() throws Exception { 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 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 9f6cfbad39e2b..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,6 +40,7 @@ public class GenericFileOnCompletion implements Synchronization { private ExceptionHandler exceptionHandler; private final GenericFile file; private final String absoluteFileName; + private GenericFileKeptDoneFiles keptDoneFiles; public GenericFileOnCompletion(GenericFileEndpoint endpoint, GenericFileOperations operations, GenericFileProcessStrategy processStrategy, GenericFile file, @@ -65,6 +66,14 @@ public void onFailure(Exchange exchange) { onCompletion(exchange); } + /** + * The done files kept for the files the poll left, when the endpoint uses a done file with a fixed name; + * null otherwise. + */ + void setKeptDoneFiles(GenericFileKeptDoneFiles keptDoneFiles) { + this.keptDoneFiles = keptDoneFiles; + } + public ExceptionHandler getExceptionHandler() { return exceptionHandler; } @@ -185,6 +194,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 && 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 0000000000000..34fd26d0b3572 --- /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. + *

+ * 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 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; + } + + @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"); + 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 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 f1cc998b316ea..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 @@ -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 --output=

` now writes one normalized file per input file into the