My NiFi pipeline fetches and processes files from an SFTP server. When the number of files to fetch does not exceed 500, everything works well. However, sometimes the number of files the pipeline has to list at once exceeds 2,000,000, which critically impacts performance. I’m trying to find a way to throttle the file stream so that no more than 500 files are processed at any given time. First Attempt: GetSFTP Processor First, I looked into the GetSFTP processor. It has a convenient property called "Remote Poll Batch Size". However, this deprecated processor has several disadvantages, the most significant being that it does not maintain state. Because of this, when I set "Delete original" to False, it fetches files infinitely. Setting "Delete original" to True is not desirable because my pipeline’s guaranteed delivery requirements mandate that files should only be deleted after successful processing (not at the start of the pipeline). Second Attempt: Queue Back Pressure Next, I explored back pressure settings in the queue between ListSFTP and FetchSFTP. For example, I set the back pressure threshold to 500 in the queue between these processors. However, as I understand it: When ListSFTP reads 2 million files, it pushes all of them into the queue. ListSFTP can only add next files to the queue once FetchSFTP has processed enough files to bring the count below the back pressure limit (e.g., after 1,900,501 files are consumed). This behavior is not what I expected. Expected Behavior: I wanted one of the following: ListSFTP lists no more than 500 files at a time, or The queue between ListSFTP and FetchSFTP holds no more than 500 files, or FetchSFTP fetches no more than 500 files at a time. Question: Is there a way to solve this problem without creating a custom processor or using Groovy scripts?
... View more
