CAMEL-25728: camel-file - keep a done file with a fixed name for the files a poll left

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]
 ====

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]
 ====

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<T> extends ScheduledBatchPollingConsum
     private boolean retrieveFile = true;
     // files already reported as resolving outside the starting directory (bounded)
     private final Map<String, Boolean> skippedOutsideStartingDirectory = LRUCacheFactory.newLRUCache(1000);
+    // the done files with a fixed name kept for the files a poll left (null if the endpoint has no such done file)
+    private final GenericFileKeptDoneFiles keptDoneFiles;
 
     protected GenericFileConsumer(GenericFileEndpoint<T> endpoint, Processor processor, GenericFileOperations<T> operations,
                                   GenericFileProcessStrategy<T> processStrategy) {
@@ -79,6 +83,10 @@ protected GenericFileConsumer(GenericFileEndpoint<T> 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<String> listedDoneFiles = startDoneFileBatch(kept, files, !listingCut);
+            if (listingCut) {
+                List<GenericFile<T>> excessive = files.subList(maxMessagesPerPoll, files.size());
+                removeExcessiveInProgressFiles(excessive);
+                excessive.clear();
+                limitHit = true;
+                listedDoneFiles.forEach(kept::keep);
+            }
+        }
+
         // log if we hit the limit
         if (limitHit) {
             LOG.debug("Limiting maximum messages to poll at {} files as there were more messages in this poll.",
@@ -201,6 +225,10 @@ 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<Object> 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<Object> 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<GenericFile<T>> files) {
         }
     }
 
+    /**
+     * The done files with a fixed name kept for the files left by the last poll, or <tt>null</tt> if not used (no such
+     * done file, or a custom processor that does not complete the files).
+     */
+    GenericFileKeptDoneFiles getKeptDoneFiles() {
+        return customProcessor == null ? keptDoneFiles : null;
+    }
+
+    /**
+     * Starts the batch of the kept done files with the files listed, and deletes the done files kept for files that are
+     * gone.
+     *
+     * @return the done files of the files listed
+     */
+    private Set<String> startDoneFileBatch(GenericFileKeptDoneFiles kept, List<GenericFile<T>> files, boolean complete) {
+        Set<String> doneFiles = new HashSet<>();
+        for (GenericFile<T> file : files) {
+            doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+        }
+        List<String> gone = kept.startBatch(doneFiles, complete);
+        if (gone != null) {
+            // the done file was kept for files that are no longer there, so it is no longer needed
+            for (String doneFileName : gone) {
+                try {
+                    boolean deleted = operations.deleteFile(doneFileName);
+                    LOG.trace("Done file: {} kept for files that are gone was deleted: {}", doneFileName, deleted);
+                } catch (Exception e) {
+                    LOG.debug("Cannot delete done file: {} due to: {}", doneFileName, e.getMessage(), e);
+                }
+            }
+        }
+        return doneFiles;
+    }
+
+    private void keepDoneFile(GenericFile<?> file) {
+        GenericFileKeptDoneFiles kept = getKeptDoneFiles();
+        if (kept != null) {
+            kept.keep(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+        }
+    }
+
     /**
      * Whether or not we can continue polling for more files
      *
@@ -340,6 +419,12 @@ public 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<T> onCompletion
+                    = new GenericFileOnCompletion<>(endpoint, operations, processStrategy, target, absoluteFileName);
+            onCompletion.setKeptDoneFiles(getKeptDoneFiles());
+            exchange.getExchangeExtension().addOnCompletion(onCompletion);
 
             LOG.debug("About to process file: {} using exchange: {}", target, exchange);
 
@@ -1011,6 +1098,9 @@ protected void doStart() throws Exception {
     protected void doStop() throws Exception {
         prepareOnStartup = false;
         skippedOutsideStartingDirectory.clear();
+        if (keptDoneFiles != null) {
+            keptDoneFiles.clear();
+        }
         super.doStop();
         ServiceHelper.stopService(processStrategy);
     }

components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileKeptDoneFiles.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.file;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * The done files with a fixed name (such as {@code doneFileName=ready}) that a consumer keeps because a poll left files
+ * of their directory for the next poll.
+ * <p/>
+ * A file is left when the poll did not take it: beyond {@code maxMessagesPerPoll}, or its processing could not start
+ * (for example the read lock was not acquired). The last file of the poll then does not delete the done file, so the
+ * next poll consumes the files left, and any other file of the directory, as it does whenever the done file is there. A
+ * file that was taken and failed is not left: as before, the done file is deleted after the last file of the poll
+ * unless that file failed.
+ * <p/>
+ * The state is in memory and cleared when the consumer stops.
+ */
+final class GenericFileKeptDoneFiles {
+
+    private final Set<String> kept = new HashSet<>();
+
+    /**
+     * Called once the directory is listed, before the files are consumed: forgets the done files kept by the last poll,
+     * as the files of this poll decide again whether they are kept.
+     *
+     * @param  listedDoneFiles the done files of the files this poll found
+     * @param  listingComplete whether every file was listed; if not (the listing stopped at the limit), a done file not
+     *                         listed is kept, as its files may be among those not listed
+     * @return                 the done files kept for files that are gone (the listing was complete and found none of
+     *                         their files), to delete
+     */
+    synchronized List<String> startBatch(Set<String> listedDoneFiles, boolean listingComplete) {
+        List<String> answer = null;
+        if (listingComplete) {
+            for (String doneFileName : kept) {
+                if (!listedDoneFiles.contains(doneFileName)) {
+                    if (answer == null) {
+                        answer = new ArrayList<>();
+                    }
+                    answer.add(doneFileName);
+                }
+            }
+            kept.clear();
+        } else {
+            kept.removeAll(listedDoneFiles);
+        }
+        return answer;
+    }
+
+    /**
+     * Keeps the done file, as this poll leaves files of its directory for the next poll.
+     */
+    synchronized void keep(String doneFileName) {
+        kept.add(doneFileName);
+    }
+
+    /**
+     * Whether the done file must be kept after the last file of the poll.
+     */
+    synchronized boolean isKept(String doneFileName) {
+        return kept.contains(doneFileName);
+    }
+
+    synchronized void clear() {
+        kept.clear();
+    }
+}

components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java

@@ -40,6 +40,7 @@ public class GenericFileOnCompletion<T> implements Synchronization {
     private ExceptionHandler exceptionHandler;
     private final GenericFile<T> file;
     private final String absoluteFileName;
+    private GenericFileKeptDoneFiles keptDoneFiles;
 
     public GenericFileOnCompletion(GenericFileEndpoint<T> endpoint, GenericFileOperations<T> operations,
                                    GenericFileProcessStrategy processStrategy, GenericFile<T> file,
@@ -65,6 +66,14 @@ public 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;
+     * <tt>null</tt> 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 {