Hi,
Not over complicating your scenario and assuming that you get a .trg file every once in a while where you dont have to worry about clashes or concurrency issue, I would solve this as follows
1- Use ListFile processor and points it to target directory. This processor needs to run on a schedule where its not continuously reading the same file while you are processing them. You have to figure out how much time between the different listing is enough to process the files in case of trg arrival. Also make sure to set the Record Writer property set so that you get an array of all the files in one flow file. there wont be tracking in this case (set listing strategy No Tracking) since we will be continuously reading the same files again and again in case no trg file has made it yet. The output of this processor is going to be an array of all the files found where each file object has the following properties (assuming we have json writer):
{
"filename": "...",
"path": "....",
"directory": false,
"size": 256496,
"lastModified": 1707490322483,
"permissions": null,
"owner": null,
"group": null
}
2- Use QueryRecord by adding dynamic property with the following query:
select * from flowfile where exists (
select 1 from flowfile where filename like '%.trg'
)
This will produce the array list from above only if .trg file is found amongs them, otherwise nothing will happen and we will wait for the next listing from above.
3- If the condition above is met and trg file has made it , then use SplitRecord (or SplitJson in case you are using json writer ) to split each file object.
4- Use EvaluateJsonPath to get the filename and path for each file object.
5- Use FetchFile provided the attributes above to get the file and then do whatever needed. Make sure to set the completion strategy to move or delete the file so that you dont re process again.
This is a very simplistic solution that might work in case like I said you get .trg file every once in a while where there is enough time to process each trg files batch. Also if you are not dealing with large number of files. If any of those conditions are not met , you definitly have to re consider. Another option that would work better , is to have two flows where one is continuously picking up whatever files come and place it in staging area and log it in the DB, so that when trg file arrives you invoke the other flow using nifi api to read and process whatever got logged in the DB. The DB table will have the staging area path for each logged file so you pass that to the FetchFile processor. This way you can manage clashes and concurrency issues better as well as you dont have to continuously keep listing all the files like above and query the dataset to look for trg files. The files already has been moved to the staing area and whenever trg arrives the list is read once and the files are processed.
If find this helpful please accept the solution.
Thanks
