Hello,
I'd like to implement a lock mechanism in a data flow that prevents a flow file from progressing. Similar in a way to the wait processor, holding a flowfile in a queue while another flowfile with a corresponding signal attribute passes through a 'gate' further down the flow, similar to a notify.
In my specific case I load a number of files received together in to target tables.
I have a single flow which does the following:
- PutSql - issues a truncate table command that empties the target staging table. The table name is taken from an attribute.
- PutDatabaseRecord - Inserts the records into the staging table based on the same attribute.
- PutSql - executes a stored procedure passing the relevant tables name as parameter from the attribute. This SP merges data from the staging table into a 'main' table.
A number of these files, share the same staging and target tables and on occasions I have a race condition where the the staging table is truncated by a second flowfile before the stored procedure has had time to run for the first.
There are never more than two flowfiles destined for the same target table currently, so I'm able to mitigate the issue by routing one set to a retry processor and penalising the flowfiles for an arbitrary amount of time to give the first set time to complete their load. This isn't particularly elegant and it doesn't feel like it would scale very well. Also, my situation is likely to change, and I could have more than two common files in the future.
To my mind, if I could have a wait processor check a cache when a flowfile passes through it, and if the cache is empty it adds the target table name value to the cache as a signal. After the third step above is complete for this flowfile, it passes through a notify type processor which removes the signal attribute from the cache releasing any flowfiles with a matching attribute from the queue on the earlier wait.
I don't think I can get the wait and notify to work in this manner, but I could be wrong? If so, is there another way to achieve this type of functionality>
I'm aware of the Processor Group FlowFile Concurrency and Outbound Policy settings, but these would be too restrictive only processing a single flow file at a time. I would like to only process as many flowfiles concurrently as the DB will take, only holdiing the flowfiles which might cause the race condition.
I hope that's clear. Thanks in advance.
