How to detect all branches in a NiFi flow have finished processing?

@iriszhuhao, Welcome to our community! To help you get the best possible answer, I have tagged our NiFi experts @MattWho @cotopaul @SAMSAL  who may be able to assist you further.

Please feel free to provide any additional information or details about your query, and we hope that you will find a satisfactory solution to your question.

Hi @iriszhuhao ,

The wait processor is supposed to be used with the Notify processor to work correctly and its more suited for when  you have a data that will be split downstream  where each split flowfile will have a common identifier and split count that the Wait-Notify processors can go by  as explained in this article:

https://pierrevillard.com/2018/06/27/nifi-workflow-monitoring-wait-notify-pattern-with-split-and-merge/

When you get the data from SQL do you get it all in one flowfile or its already split where each record in one flowfile. If you are not getting a lot of data then I would recommend to get it all in one flowfile and then split it so that you can use this pattern and to run the final SP once. So the design will look something like this:

In this case the Wait-Notify Signal identifier will be the value of the fragment.identifier and the Wait Target Signal Count will be the value of the fragment.count which both written by the Split processor.

If you find this helpful please accept solution.

Thanks

@iriszhuhao, to be really honest, your problem is not 100% related to NiFi, as this is more of an architectural question.

How would you achieve what you are trying to achieve in another ETL Graphical Tool (Informatica, Talend, Knime, etc)? That logic should be applied in your case as well, as most of the tools perform the same thing but in a different manner. 

Now, going back to your question, bear in mind that this might not be the best solution, but you could implemented a signaling table where you store the execution of each of your 9 flows. Basically you create a database table in which you will store something like PROCESSOR_ID, PROCeSSOR_NAME, EXECUTION_TIME, SUCCES. From each of your success flows, you will go into a processor where you generate this data and afterwards insert the content into your table. Next, out of your final processor (the one saving the data in your database) you could into an RetryFlowFile where you set a penalty of X seconds/hours/minutes. From that flow file, you then go into an ExecuteSQLRecord and interogate your table. The output of the ExecuteSQLRecord is then sent into a query processor and check if all 9 flows have entered a row in your table. If all 9 inserted a row in your table, you proceed to do what you want. If not all 9 flows inserted a row in your table, you go into an processor where you call the NiFi's Rest API to check the status of your affected processors (you can do that with an ExecuteStreamCommand, ExecuteScript, InvokeHTTP, etc). You can call the /nifi-api/flow/processors/YOUR_MISSING_PROCESSOR/status API and check the results (statsLastRefreshed, flowFilesIn, flowFilesOut) to see whether something came or should have came through that flow. Next, based on your logic, you can proceed next to checking other processors or call the stored procedure to end your flow. Lastly, you can truncate your table once your execution is done, if you do not want to keep some historical data there. 

Again, this might not be the best solution, but this is the first thing I would try if I were in your situation. 

I had a need for multiple Lookups...custom Groovy processor with several LookUp services as a part of it...consolidated that, routed accordingly, and performed faster.