max-messages-per-poll and max-fetch-size not working as expected

[SimoneGiusso]

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

6.2.4

Describe the bug

According to the documentation the following combination of values max-messages-per-poll=-1 and max-fetch-size=1 should have the following effect: the sftp synchronizer fetch all the files from sftp server but one at time. As soon as a file is downloaded, a message with the file as payload, is fired in the channel.

However with this setting, in a single pool, the synchronizer only fetch 1 file.

To Reproduce

Sftp server containing more than 1 files and max-messages-per-poll=-1 and max-fetch-size=1

Expected behavior

Fetch all the files and sent a message as soon as a file is downloaded.

[artembilan]

Sorry, I need more explanation what is expected.
The max-fetch-size=1 means only transfer to local directory one of per fetch:

filteredFiles = filteredFiles.stream()
					.limit(maxFetchSize)
					.collect(Collectors.toList());

Now the max-messages-per-poll=-1 is like this:

					while (this.initialized && (this.maxMessagesPerPoll <= 0 || count < this.maxMessagesPerPoll)) {
						if (this.maxMessagesPerPoll == 0) {
							logger.info("Polling disabled while 'maxMessagesPerPoll == 0'");
							break;
						}
						if (pollForMessage() == null) {
							break;
						}
						count++;
					}

That pollForMessage() calls:

	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, as long as you ask for only one file to synchronizeToLocalDirectory there is no way to cache more of them in the local dir.

Would you mind to elaborate more what would you like to see?

[SimoneGiusso]

I set the max-messages-per-poll (infinity) to a higher value than max-fetch-size (1). Now when max-messages-per-poll > max-fetch-size this should be the behaviour, according to the documentation:

max-messages-per-poll=2 and max-fetch-size=1: The adapter fetches one file, emits it, fetches the next file, and emit it. Then it sleeps until the next poll.

According to the description, with my configuration, I would expect that the adapter fetches one file, emits it, fetches the next file, emit it, etc... until all the files have been fetched. However the behaviour I'm seeing is the following:

adapter fetches one file, emits it

That's it!!! No more files are downloaded to the local repository, hence no more messages are emitted.

[artembilan]

No more files are downloaded to the local repository, hence no more messages are emitted.

Then something is off with the filter for that channel adapter.
The logic there is like this:

			boolean filteringOneByOne = haveFilter && this.filter.supportsSingleFileFiltering();
			List<F> filteredFiles = applyFilter(files, haveFilter, filteringOneByOne, maxFetchSize);

May we have your whole configuration for that channel adapter to investigate, please?

[SimoneGiusso]

It's not the filter. If I don't' set the max fetch size all the files are downloaded

[artembilan]

I cannot reproduce it in my testing environment:

	@Test
	public void testSftpInboundFlow() {
		QueueChannel out = new QueueChannel();
		IntegrationFlow flow = IntegrationFlow
				.from(Sftp.inboundAdapter(sessionFactory())
								.preserveTimestamp(true)
								.remoteDirectory("sftpSource")
								.regexFilter(".*\\.txt$")
								.localFilenameExpression("#this.toUpperCase() + '.a'")
								.localDirectory(getTargetLocalDirectory())
								.remoteComparator(Comparator.comparing(SftpClient.DirEntry::getFilename))
								.maxFetchSize(1),
						e -> e.id("sftpInboundAdapter").poller(Pollers.fixedDelay(100).maxMessagesPerPoll(-1)))
				.channel(out)
				.get();
		IntegrationFlowRegistration registration = this.flowContext.registration(flow).register();
		Message<?> message = out.receive(10_000);
		assertThat(message).isNotNull();
		Object payload = message.getPayload();
		assertThat(payload).isInstanceOf(File.class);
		File file = (File) payload;
		assertThat(file.getName()).isEqualTo(" SFTPSOURCE1.TXT.a");
		assertThat(file.getAbsolutePath()).contains("localTarget");

		message = out.receive(10_000);
		assertThat(message).isNotNull();
		file = (File) message.getPayload();
		assertThat(file.getName()).isIn("SFTPSOURCE2.TXT.a");
		assertThat(file.getAbsolutePath()).contains("localTarget");

		registration.destroy();
	}

Both files are downloaded as expected: one by one.

Apparently you indeed need to share with me some simple project to reproduce.

Thanks

[SimoneGiusso]

@artembilan here you are -> https://github.com/SimoneGiusso/spring-integration-sftp-demo/tree/main

The test run against a real sftp server. The test should not pass since only one file is downloaded instead of the two available which satisfy the filter.

I also noticed this warning:

2024-05-07T22:18:10.611+02:00  WARN 79505 --- [   scheduling-1] o.s.i.s.i.SftpInboundFileSynchronizer    : The remote file 'File2_data.csv[-rw-r--r--    1 501      dialout         0 May  7 20:17 File2_data.csv]: type=1;size=0;uid=501;gid=20;perms=0x81a4;flags=[Size, UidGid, Perms, AccessTime, ModifyTime];owner=null;group=null;aTime=2024-05-07T20:18:10Z;cTime=null;mTime=2024-05-07T20:17:11Z;extensions=[]' has not been transferred to the existing local file '/tmp/extractions/File2_data.csv'. Consider removing the local file.
2024-05-07T22:18:10.611+02:00  WARN 79505 --- [   scheduling-1] o.s.i.s.i.SftpInboundFileSynchronizer    : The remote file 'File1_data.csv[-rw-r--r--    1 501      dialout         0 May  7 20:17 File1_data.csv]: type=1;size=0;uid=501;gid=20;perms=0x81a4;flags=[Size, UidGid, Perms, AccessTime, ModifyTime];owner=null;group=null;aTime=2024-05-07T20:18:10Z;cTime=null;mTime=2024-05-07T20:17:11Z;extensions=[]' has not been transferred to the existing local file '/tmp/extractions/File1_data.csv'. Consider removing the local file.

I don't think is expected. Something is wrong.

[artembilan]

I think the problem is really here:

fileSynchronizer.setFilter(SftpSimplePatternFileListFilter("*_data.csv"))

You have to compose it with once filter , like we do in the framework for DSL:

spring-integration/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/dsl/SftpInboundChannelAdapterSpec.java

Line 74 in 8b88668

private CompositeFileListFilter<SftpClient.DirEntry> composeFilters(FileListFilter<SftpClient.DirEntry>

[SimoneGiusso]

I don't understand. The filter is not applied to each file to decide which one to download? Why max-fetch-size influences the behavior of the filter?

Why when max-fetch-size is not set, the test pass?

[artembilan]

Yes, that’s the right question. I cannot run your application since I don’t have real SFTP server on my Windows. I need to develop something similar to be able to reproduce. I guess my embedded MINA should do the trick. The point is that the problem is somewhere in app or in the framework as you claim.

[SimoneGiusso]

I think it should work on windows as well. The ssh server is run inside a docker container. You shouldn't doing anything. TestContainer should pull and run the image for you.

Please tell me if I can help with the errors you are facing when running the demo