Shouldn't list remote file server each time max-fetch-size is reached

[karim789]

In what version(s) of Spring Integration are you seeing this issue?

all versions up to 6.5.x

Describe the bug

AbstractInboundFileSynchronizingMessageSource shoud list the remote server only once per poll.
However what happens here is each time you download max-fetch-size files, and max-message-per-poll isn't reached, then instead of working on the initial list until, it will list the remote server again, which is unnecessary and costs a lot of time.

Currently if you need a jdbc transaction per file downloaded and keep it to the end of the message flow, you need to use max-fetch-size=1, but then you end up having the remote server listed before downloading each single file, which slow down things considerably.

Chatgpt told me that AbstractRemoteFileStreamingMessageSource could be a solution to this, because it was maintaining a queue that was filled from remote file server list only when queue is empty.
But here I see the queue is cleared when maxFetchSize is reach, therefore it will trigger a remote file server list right after that

		if (maxFetchSize > 0 && this.fetched.get() >= maxFetchSize) {
			this.toBeReceived.clear();
			this.fetched.set(0);
		}

(

spring-integration/spring-integration-file/src/main/java/org/springframework/integration/file/remote/AbstractRemoteFileStreamingMessageSource.java

Line 219 in 2315423

if (maxFetchSize > 0 && this.fetched.get() >= maxFetchSize) {

)

Therefore I see no solution to have a transaction per downloaded file and it's emitted message without having bad performances.

To Reproduce

Use a max-fetch-size=1

Expected behavior

Work on the remote file list until exhausted, before polling the server for a file list again.

Sample

Uncessary as code analysis shows the issue.

[artembilan]

Can we understand why ChatGPT is involved here? Does it really give you a solution?

What is initial list in your meaning, since the reason of max fetch is to not fetch from remote server more.

We will look in details tomorrow.

[karim789]

Can we understand why ChatGPT is involved here? Does it really give you a solution?

I wanted to credit chatgpt for pushing me to look into AbstractRemoteFileStreamingMessageSource, but after writting the post I realised that although chatgpt seemed to be aware that the file list was maintened with a blockingqueue in the message source, it didn't saw that it was cleared after max-fetch-size was reached.
So no it didn't really gave a solution at all, and I could have scrapped this part in the message.

What is initial list in your meaning, since the reason of max fetch is to not fetch from remote server more.

If we use the concept of polling, and set a periodic polling, I would except to have only one interrogation of the remote source for a polling cycle, and then work only on this datas for the polling cycle.

I don't think max-fetch-size should implicate listing the remote server again in the same polling cycle, because stuff that would require to list the remote server again that I think of, like get the most up to date files, could be achieved by using a shorter polling period and a smaller max-messages-per-poll.

So I see the max-fetch-size more as a way to deal with other stuff, like having the possiblity aggregate a batch of files in the same spring integration flow, or work with a set of files in the same database transaction.

And on that mater my observation is that the transaction is commited after the first message is sent to the outbound target, so this is a bit weird, because you commit the metadatas for the batch of 10 downloaded files only when the first message is sent, and eventually rollback everything if it fails.
But if there is a failure on the second message, then metadatas for the 10 messages are already commited through first message success, and this is a bit inconsistent.

So only way to have consistency with transactions here is either to aggregate the then messages in 1, to have a commit for all metadatas, or work on max-fetch-size=1

So far globaly I struggle to have a process resilient to infrastructure errors like unavailable outbound server, credential changes, wild dns changes, brutal crashes caused by OS problems like outof memory, reboots etcetera.
The goal would is to have something than can recover without human maintenance and history cleaning.

I was having that with max-fetch-size=1 and transactions, but the systematic remote server listing kills the performances.

I tried also parallelisation with taskexecutor on the poller, but then saw that concurrency was going to be an issue with RotatingServerAdvice.

[artembilan]

I see the max-fetch-size more as a way to deal with other stuff, like having the possiblity aggregate a batch of files

This is not correct assumption.
While we try to make our components flexible and working together (like splitter+aggregator), sometime it is just not possible, and we have to deal with result of one component in another as it was from a black box for us.
The point of messages is to have a system as loose-coupled as possible.

Looks like you are mixing a lot of concern in your solution which has their own caveats.
In the end you can just implement your own MessageSource to keep everything you need as close a possible and as robust as you see that.

I'm not sure what to get from your current mix of transactions and RotatingServerAdvice.
But if you can point to something specific as you have started with a max-fetch-size that would be great.
Another idea is to raise a Pull Request with what you think could be fixed in the framework.

Thank you!

[karim789]

But we have to know what is real choices and what are side effects of adding functionalities over time.

For instance from looking at the code, the reason that RotatingServerAdvice won't work with taskscheduler is because the remotedirectory is an instance variable in the messagesource, and that multiple threads would overwrite the variable.
It seems fair to want to have multiple directories, but then this become orthogonal to multithreads, this doesn't really makes sense besides the difficulty and required time to have to refactor it.

About transactions, it seems fair that to want to store the metadata, there is the jdbcmetadatastore for that, but it's also fair to want to ensure to not have inconsistent data by using transactions.

About max-fetch-size as I said I don't really see the use case for this if it implies listing remote server for each fetch group, because the relisting use case should be covered through the periodicity or rate of the polling.

I don't think I am mixing concerns, as I am just using functionalities provided by spring integration, but some of them do not work well together because of the implementation.

I think maybe a solution would be to have a PollingCycleContext or PollingCycleSession that would travel through methods parameters, or maybe ThreadLocal bound, that would store the file list, the folders on which to rotate on, stuffs like that, otherwise I think the current implementation can't be fixed by just overriding a few thing there and there. Some methods are private, some are protected, some public, you don't really know why, but hey that's normal, I would not do better.

So yeah of course I would be glad that the company I use spring integration for, let me do big modifications, or that they pay you, but even currently if I modify it on my work time on the company laptop, I can't even push it to github because it's write blocked. I heard that maybe some demands could be made for some teams, I will try to see if there is a way to submit PR in someway, because we use a corporate github.

Hopefully I can work in a more open company someday, but it's the kind that use opensource for free and doesn't want to contribute anything at all.

But what I really need is something resilient to crashes and datas stored in a database, and performant enough, and I was close to that if the max-fetch-size when reach wasn't clearing the list.

[artembilan]

I understand your struggle.
Probably all those additions to those components were added for some specific use-cases without taking into account some others.

Let's see if we can dig out something with you one by one!
Even if you cannot contribute via PRs, you still can post some links and code snippets over here in our discussion.

So, first of all re. RotatingServerAdvice.
Yes, it is indeed to not good to be used with concurrent threads.
Just because it does mutate the target MessageSource as you have already pointed out.

How about instead of this RotatingServerAdvice against a single AbstractInboundFileSynchronizingMessageSource, have several instances for that source for different remote directories?
This way no any extra logic to rotate dirs and those dirs are going to be polled in parallel.

Looks we indeed can only advertise RotatingServerAdvice when a limited number of threads can be used in the application, so we cannot have several polling tasks in parallel, but have to rotate over directory list in only one of them.

I'll look max-fetch-size a bit later.

I still cannot get your transaction and metadata concern.
Can you elaborate more what exactly do you mean?
Thanks

[karim789]

How about instead of this RotatingServerAdvice against a single AbstractInboundFileSynchronizingMessageSource, have several instances for that source for different remote directories?
This way no any extra logic to rotate dirs and those dirs are going to be polled in parallel.

For this source, the remote directories are each a directory created dynamically each day with the current date with file that can be added in older folders, so we want to keep open to parametrize through application.yaml at what date depth we probe directories.
So this mean we would need to dynamically create flows, and spring beans, which is a bit more tricky, and less maintainable, I tried it for other stuff and don't want to rely to much on that.

By the way up until now I used an ExpressionFunction to handle rotation of the folder name, because I was discovering the framework and StandardRotationPolicy didn't allow for dynamic foldernames, and recently I figured how to reimplement RotationPolicy, but that didn't really help.

I think the RotationPolicy tries kill to birds with one stone without departing of the idea of a unique remote directory to probe, while probably it would be better to have something like a RemoteFileScanStrategy to build the list of files to download, that could implements either single directory, list of directory, recursivity.
Even fairness and order could be handled in the strategy.

Also another issue I faced by the way is that the object returned by the sftpclient do not contain the fullpath of the file, only the name, therefore since files have the same name in each date folder, this is an issue with the SftpPersistentAcceptOnceFileListFilter that cannot differenciate files from different directories.

	@Override
	protected String fileName(SftpClient.DirEntry file) {
		return file.getFilename();
	}

spring-integration/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java

Line 49 in 0386291

return file.getFilename();

The way I got around that at first was that I added the whole source code of spring-integration-sftp and wrapped the SftpClient.DirEntry in an object that contained the parent folder.
I improved that a bit by only duplicating a few classes like SftpSession so I could use sftpClient.scandir( ) instead of sftpClient.readdir( ), because it returns a SftpClientDirectoryScanner.ScanDirEntry that extends DirEntry and has a getFullPath(), and just cast to it when I need

spring-integration/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java

Line 127 in 0386291

return StreamSupport.stream(this.sftpClient.readDir(remoteDir).spliterator(), false)

I still cannot get your transaction and metadata concern.
Can you elaborate more what exactly do you mean?

I needed to persist metadata to not perpertually download the same files, so I used PropertiesMetadataStore.
I noticed at this time that in case of application crash or wild server restart, you could have files that were staying in the localfolder, so I suspected issues where they were never sent.
There is also the case of unavailable outbound gateway were the flow fail, and about that I didn't tried yet the retyadvice, but there is probably something to work on here.

So anyway the point is that in some cases I can have a failed file process for various reasons, and so far it feels like the best way to solve it, is to clean the various metadata refering to that file in the metadatastore, and let the process start clean again.

Since we have DRP servers, in case we need to switch to the backup server we needed to copy the metadataproperties to the other server, so that's here that it made sense to rely on a database and switch to JdbcMetadataStore, which would make also easier to clean inconsistent state due to failure, and also with the promise that I could leverage transaction automatic rollback to clean the metadata for the current file process and start the process over in case something went wrong.

Also there is something that maybe I went in the wrong direction thinking transactions could help, it's about how the synchroniser first download files in the local folder if it's empty, and put in the store metadata about remote URI of the downloaded files.
Then the LocalFileReader that build the actual messages, scans this localfolder and do metadatastore.get(fileKey) to gather metadata of the file from the URI parsing and create header like REMOTE_DIRECTORY FILE_NAME etcetera to continue the process.
So if you have a file in there without the according metadata and if you need the REMOTE_DIRECTORY header in a later stage, then it will stay here and fail infinitely, that's why I try to delete files after success of failure, but I think here there is something I am not handling correctly.

So the only thing I found that worked was to clean this local folder on startup, clean the succes and failed file in this folder as well I can using either channel interceptor or spring integration precommit postcommit and rollback transaction interface, along the use jdbc and spring transaction with jdbcmetadatastore, and have a transaction per file, wich requires to have max-fetch-size=1

And this is were I ended up realising performances where too slow because max-fetch-size=1 was listing remote server before downloading each file.

Thanks

Thank you too !

[artembilan]

@karim789 ,

sorry for delay: was busy with releases this week.
And thank you for comprehensive investigation!

For this source, the remote directories are each a directory created dynamically each day

I think that could be done via AbstractInboundFileSynchronizer and its:

	/**
	 * Specify an expression that evaluates to the full path to the remote directory.
	 * @param remoteDirectoryExpression The remote directory expression.
	 * @since 4.3.13
	 * @see #setRemoteDirectoryExpression(Expression)
	 */
	public void setRemoteDirectoryExpressionString(String remoteDirectoryExpression) {

So, you have a bean which do some trick on the remote directory name to return something relevant at the moment when the synchronizeToLocalDirectory() is called.

Also another issue I faced by the way is that the object returned by the sftpclient do not contain the fullpath of the file

This is really a problem.
Please, raise a new GH issue and we will think what to do.
We can talk about details (including ScanDirEntry) over there, in that specific context.

So if you have a file in there without the according metadata

You probably mean something like your application has been restarted and you have files in the local dir from the previous run.
Yes, it is indeed could happen if we rely on a default, in-memory SimpleMetadataStore .
You can change it to something like PropertiesMetadataStore. This probably has to be co-located with an application instance and really be relevant only to this application instance, because other instances has their own local directory on the matter.

If we use the concept of polling, and set a periodic polling, I would except to have only one interrogation of the remote source for a polling cycle, and then work only on this datas for the polling cycle.

I'm not sure what you are talking about here.
There logic there is like this:

	public final AbstractIntegrationMessageBuilder<File> doReceive(int maxFetchSize) {
		AbstractIntegrationMessageBuilder<File> messageBuilder = this.fileSource.doReceive();
		if (messageBuilder == null) {
			this.synchronizer.synchronizeToLocalDirectory(this.localDirectory, maxFetchSize);
			messageBuilder = this.fileSource.doReceive();
		}

So, it calls the synchronizer only if LocalFileReadingMessageSource returns null.
Therefore, all the fetched data is exhausted and we have not reached max-message-per-poll.
That's really expected and I believe even mentioned in the docs that remote server might be fetched several times during one poll to fulfill ``max-message-per-pollexpectation. If you are talking aboutAbstractRemoteFileStreamingMessageSource` implementation, then let's don't mix concerns and be specific.

So only way to have consistency with transactions here is either to aggregate the then messages in 1, to have a commit for all metadatas, or work on max-fetch-size=1

I see what you mean about transactions. We run one around MessageSource.receive(). During that process we do fetch and filtering, indeed this last one goes as a whole list of records in the MetadataStore.
However, in the end we still return just one file for downstream processing.
And you are right, if the second one fails, then its transaction rollback does nothing because the metadata was committed by the first one.
Not sure what to do since that is how it was designed, probably without transactions in between in mind...
We definitely cannot batch them since that is not this channel adapter goal.
But we probably can think about having a dedicated transaction on that JdbcMetadataStore.
This way we would always commit everything we have in hands to pursue original task for this fetch synchronization - performance optimization with less remote server calls.
The message processing and its own transaction really has nothing to do with metadata store.
Does that make sense?
And we can repurpose this issue for exactly this transaction "abuse".
Or do I miss anything else?

[karim789]

I think that could be done via AbstractInboundFileSynchronizer and its:

Yes that's what I did with setRemoteDirectoryExpression(Expression) for months.
I realised only recently that I could use StandardRotationPolicy as a basis for implementing my own RotationPolicy.

Please, raise a new GH issue and we will think what to do.
ok

You probably mean something like your application has been restarted and you have files in the local dir from the previous run.
Yes, it is indeed could happen if we rely on a default, in-memory SimpleMetadataStore . You can change it to something like PropertiesMetadataStore. This probably has to be co-located with an application instance and really be relevant only to this application instance, because other instances has their own local directory on the matter.

I used PropertiesMetadataStore since the begining and now use JdbcMetadataStore.
Since I added transactions in the mix, I saw other side effects.

For instance you use maxfetchsize=10, the transaction will include the 10 metadatastore.put( ), some metadata I have to put to for a specific business logic, and commit after the first message is sent.
In case there is a failure on first message, then the 10 metadatastore.put( ) will be reverted, as well as my metadata business logic infos, and you end up with 10 files without the required metadata.

I will try to find a solution without transactions, if I can't have transactions that can englobate 10 messages, or if maxfetchsize=1 list folder for all files to download.

So, it calls the synchronizer only if LocalFileReadingMessageSource returns null. Therefore, all the fetched data is exhausted and we have not reached max-message-per-poll. That's really expected and I believe even mentioned in the docs that remote server might be fetched several times during one poll to fulfill ``max-message-per-pollexpectation.

Yes but the right behavior should be to probe the state of the server once to obtain a file list, then you can do batches of files with max-fetch-size.
But you should need to list the remote server each time.
If you think listing the server each time helps to reach max-messages-per-poll, then this is rather incorporating a min-messages-per-poll behavior.
Listing only once in the begining should be enough to fullfill a constraint of max number of message to not exceed.

With the current behavior, imagine you want to download 10G files from a remote server that has a folder with 1000 files in it, but have only a local storage of 15G for processing.
This means that you will need to have a max-fetch-size of 1, because you can't have more than one file locally.
The issue is that each time you will pay the cost of listing 1000 files.

[artembilan]

I'm not sure why you have closed the issues, but OK.

but the right behavior should be to probe the state of the server once to obtain a file list, then you can do batches of files with max-fetch-size.

So, what you mean it to fetch from the server whatever is there, store somewhere in memory and then download only max-fetch-size.
Then its purpose is eliminated. It is exactly about a fetch.
What is the server state has been changed in between?
The goal of max-fetch-size is really to come for some optimal behavior to keep the whole system state as clean as possible.
That's why we could fetch server again during the same poll if max-fetch-size is exhausted and max-message-per-poll has not been reached yet.
You might think about not using max-fetch-size with the meaning to download everything is there for a single poll cycle.
Yes, then local storage limit is an issue for big files.
Then the max-message-per-poll should do the trick to remove local files when processing is done.
With that a max-fetch-size could be exactly the same what is max-message-per-poll. So, you would not exceed more than one server call if you have a lot of files.
If you do not have so many, then it should be OK to wait for more sever calls.
You may consider to use a ReceiveMessageAdvice with its conditional beforeReceive() implementation.
And afterReceive() to return just that result argument.
In case of null result, the polling loop is going to be broken and we exit.

I will try to find a solution without transactions

I think this local files synchronization is not very good for transactions.
When transaction fails we have to remove local files copies as well, not just their metadata.
That might be an issue right now in the AbstractInboundFileSynchronizingMessageSource and its AbstractInboundFileSynchronizer counterpart.
Feels like an AbstractRemoteFileStreamingMessageSource would fit more to transactions and big files.

WDYT?

[karim789]

I'm not sure why you have closed the issues, but OK.

A bad clicking I think, I didn't wanted to close it.

[artembilan]

it list the remote server that maxfetchsize is useless as it

Well, not really. As I said in my post that we call server again when it is exhausted.
And if we remove local files after processing, we don't overhead storage.
If we don't limit (by default) all available remote files at that moment are copied to local files system, and that's where we might have problem we big files.
We really might have problem right now after our investigation that this remote-to-local-copying is not compatible with transactions.