allthingssecurity commented on code in PR #28035:
URL: https://github.com/apache/camel/pull/28035#discussion_r4306582766
##########
components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java:
##########
@@ -323,15 +353,51 @@ protected void
removeExcessiveInProgressFiles(List<GenericFile<T>> files) {
}
}
+ /**
+ * The files left by the last poll while a done file with a fixed name was
there, or <tt>null</tt> 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<GenericFile<T>> files) {
+ Set<String> doneFiles = new HashSet<>();
+ for (GenericFile<T> file : files) {
+
doneFiles.add(endpoint.createDoneFileName(file.getAbsoluteFilePath()));
+ }
+ List<String> 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
*
* @param fileList the current list of gathered files
* @return <tt>true</tt> to continue, <tt>false</tt> 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) {
Review Comment:
Good catch. Two changes in e47aeb82b34e:
- Cost: the consumer no longer lists the whole directory. With the eager
limit it now lists one file past `maxMessagesPerPoll`, which is enough to know
that the poll leaves files. So a `pollEnrich` / `ConsumerTemplate.receive`
lists at most 2 files, where main lists 1. 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. That keeps `recursive=true` safe when the
listing stops in another directory.
- Test and upgrade guide:
`testConsumerTemplateReceivesEveryFileOfTheDoneFile` writes a, b, c and
`ready`, calls `consumer.receiveBody(..."?doneFileName=ready", 5000)` three
times, gets A, B and C, and then `ready` is deleted. On main the second receive
returns null (`expected: <[A, B, C]> but was: <[null, C]>`). The 4.24
upgrade-guide entry now says that `pollEnrich` and `ConsumerTemplate.receive`
keep the done file until the last file is received, where before the first
received file deleted it.
The FTP/SFTP consumer-template, pollEnrich, done-file and maxMessagesPerPoll
ITs pass (24 ITs plus `RemoteFileIgnoreDoPollErrorTest` and `SftpPreSortTest`).
_Claude Code on behalf of allthingssecurity_
##########
components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java:
##########
@@ -259,6 +286,8 @@ 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
+ doneFileLeft(getExchangeFileProperty(exchange));
Review Comment:
Right, thanks. In e47aeb82b34e `processBatch` takes the exchange file before
`processExchange`/`customProcessExchange` and keeps the done file of that file
when the processing does not start. I also took the `keepDoneFile` call out of
`removeExcessiveInProgressFiles`, because for the not-started exchanges it read
the pre-moved file again. The files cut by the limit, and those not processed
when the batch stops, are now marked explicitly with their polled file.
New test `testDownloadFailingAfterPreMoveKeepsTheDoneFileOfTheDirectory`: it
uses a `FileOperations` whose download of a.txt fails, with `preMove=work`, and
a `work/ready` file that belongs to someone else. b and c are consumed, `ready`
of the polled directory is deleted once a.txt is gone, and `work/ready` is left
alone. On the previous commit the next poll deleted `work/ready`
(`premove/work/ready should exist`).
_Claude Code on behalf of allthingssecurity_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
