Spring integration stream SFTP: pattern filter not working

Question

I have the following initial spring integration flow:

   final var inboundStreamingAdapter = Sftp.inboundStreamingAdapter(new SftpRemoteFileTemplate(sftpSessionFactory))
            .patternFilter("*.csv")
            .remoteDirectory(sftpRemoteDirectoryDownload);

   IntegrationFlows.from(inboundStreamingAdapter,
            c -> c.poller(Pollers.trigger(new PeriodicTrigger(10)).maxMessagesPerPoll(1)))
            .channel("data")
            .split(mySplitter)
            .get();

and I have the following files on my sftp server:

.DS_Store
PL_12002_1815_1.csv1

From my understanding adapter should filter out everything what does not end with csv postfix.

What is actually happening is that some of the files (.DS_Store) are sometimes making it through to mySplitter depending what is currently stored on FTP.

As far I can tell problem is in AbstractRemoteFileStreamingMessageSource (used as base class for SftpStreamingMessageSource class):

           if (this.filter != null && this.filter.supportsSingleFileFiltering()
                       && !this.filter.accept(file.getFileInfo())) {  //HERE APPLY A FILTER AND FIND OUT PL_12002_1815_1.csv1 SHOULD NOT BE PROCESSED

                   if (this.toBeReceived.size() > 0) { // don't re-fetch already filtered files 
                       //HERE WE LOAD ANOTHER FILE FROM THE QUEUE (.DS_Store)
                       file = poll();
                   }
                   else {
                       file = null;
                   }
           }
    //FILTER IS NOT REAPPLIED FURTHER DOWN AND WE CREATE A MESSAGE FOR .DS_Store
           if (file != null) {
               try {
                   String remotePath = remotePath(file);
                   Session<?> session = this.remoteFileTemplate.getSession();
                   try {
                       return getMessageBuilderFactory()
                               .withPayload(session.readRaw(remotePath))
                               .setHeader(IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE, session)
                               .setHeader(FileHeaders.REMOTE_DIRECTORY, file.getRemoteDirectory())
                               .setHeader(FileHeaders.REMOTE_FILE, file.getFilename())
                               .setHeader(FileHeaders.REMOTE_HOST_PORT, session.getHostPort())
                               .setHeader(FileHeaders.REMOTE_FILE_INFO,
                                       this.fileInfoJson ? file.toJson() : file);
       }

I would say this is a bug but maybe I just misconfigured something, can someone please help?

Answer 1

Looks like a bug; the if should be a while (with a null check).

Please open an issue on GitHub.