NIFI-190: Initial commit of Wait and Notify processors #1329
Conversation
|
Reviewing... |
There was a problem hiding this comment.
Tested this out and working nicely, a couple of minor comments.
Also, I've been trying to think if there is a simple way to support waiting for N signals. The Wait processor could have a property like "Wait Count" which would default to 1, but the Notify side seems more complicated... You either have to get all the keys from the cache and count the number of keys with some pattern, or the value of the key would have to be a count that could be atomically incremented which then gets in the way of using the value as the serialized attributes. Curious if you had any other ideas around this.
|
|
||
| public static final Relationship REL_SUCCESS = new Relationship.Builder() | ||
| .name("success") | ||
| .description("A FlowFile with a matching release signal in the cache will be routed to this relationship") |
There was a problem hiding this comment.
Minor point, but should this be something like "All FlowFiles where the release signal has been successfully entered in the cache will be routed to this relationship"
There was a problem hiding this comment.
Technically it would be: "The first FlowFile with a matching release signal in the cache...", since Wait removes the cache entry once it has successfully transferred the flow file to success.
Do you think there should be a use case where the cache value is not removed? If so, we could add a property like "Remove Successful Signals?"
There was a problem hiding this comment.
I was mostly just thinking about the wording... since this is the success relationship of Notify, doesn't it not matter what is in the cache? As long as this processor successfully puts the entry in the cache the flow file will go to success, so I was getting hung up on the words "with a matching release signal".
There was a problem hiding this comment.
Oh whoops, I must have just copied that from Wait. Thanks!
| if (logger.isDebugEnabled()) { | ||
| logger.debug("No release signal yet for {} on FlowFile {}", new Object[] {cacheKey, flowFile}); | ||
| } | ||
| session.transfer(flowFile, REL_WAIT); |
There was a problem hiding this comment.
Since the wait relationship is going to be looped back to itself and continually retrying, I'm wondering if there should be some way to throttle how fast it is retrying? The only thing I can think of would be to penalize the flow file, but then we are using the penalty duration for two different scenarios, but maybe that is fine. Thoughts?
There was a problem hiding this comment.
My thought was that you'd configure the Schedule duration according to how fast you wanted it to be retried. Do you think there should be an additional restriction?
There was a problem hiding this comment.
After I wrote this comment I also started thinking about using the scheduling to control it, that seems good to me, so no need to do anything else here.
|
I like the idea about waiting for N signals. I think it can be done by adding a getKeys(Pattern) method to the distributed cache, so maybe it's best captured in a separate ticket. |
|
That makes sense, I think we can add that functionality later. I created this JIRA to track the changes for the DistributedMapCacheClient https://issues.apache.org/jira/browse/NIFI-3214, then we can eventually use that to determine if there are N keys in the cache matching a pattern, and release based on that. I'm a +1 on this PR as is and will merge shortly, thanks for reviving this ticket! |
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
For all changes:
Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
Has your PR been rebased against the latest commit within the target branch (typically master)?
Is your initial contribution a single, squashed commit?
For code changes:
For documentation related changes:
Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.