This is an automated email from the ASF dual-hosted git repository.
pvillard pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new d049bca NIFI-8205: Documentation improvements for the Wait processor
d049bca is described below
commit d049bca7948227cdddcd59a50c2daf104e2883f8
Author: Peter Gyori <[email protected]>
AuthorDate: Fri Feb 5 18:49:39 2021 +0100
NIFI-8205: Documentation improvements for the Wait processor
Signed-off-by: Pierre Villard <[email protected]>
This closes #4808.
---
.../org/apache/nifi/processors/standard/Wait.java | 39 +++++---
.../additionalDetails.html | 110 +++++++++++++++++++--
2 files changed, 127 insertions(+), 22 deletions(-)
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/Wait.java
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/Wait.java
index 37f4479..83ac2e8 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/Wait.java
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/Wait.java
@@ -72,9 +72,15 @@ import static
org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.REJE
@InputRequirement(Requirement.INPUT_REQUIRED)
@CapabilityDescription("Routes incoming FlowFiles to the 'wait' relationship
until a matching release signal "
+ "is stored in the distributed cache from a corresponding Notify
processor. "
- + "When a matching release signal is identified, a waiting FlowFile is
routed to the 'success' relationship, "
- + "with attributes copied from the FlowFile that produced the release
signal from the Notify processor. "
- + "The release signal entry is then removed from the cache. Waiting
FlowFiles will be routed to 'expired' if they exceed the Expiration Duration. "
+ + "When a matching release signal is identified, a waiting FlowFile is
routed to the 'success' relationship. "
+ + "The release signal entry is then removed from the cache. "
+ + "The attributes of the FlowFile that produced the release signal are
copied to the waiting FlowFile if the Attribute Cache Regex "
+ + "property of the corresponding Notify processor is set properly. "
+ + "If there are multiple release signals in the cache identified by
the Release Signal Identifier, "
+ + "and the Notify processor is configured to copy the FlowFile
attributes to the cache, "
+ + "then the FlowFile passing the Wait processor receives the union of
the attributes of the FlowFiles "
+ + "that produced the release signals in the cache (identified by
Release Signal Identifier). "
+ + "Waiting FlowFiles will be routed to 'expired' if they exceed the
Expiration Duration. "
+ "If you need to wait for more than one signal, specify the desired
number of signals via the 'Target Signal Count' property. "
+ "This is particularly useful with processors that split a source
FlowFile into multiple fragments, such as SplitText. "
@@ -88,8 +94,9 @@ import static
org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult.REJE
@WritesAttribute(attribute = "wait.start.timestamp", description =
"All FlowFiles will have an attribute 'wait.start.timestamp', which sets the "
+ "initial epoch timestamp when the file first entered this processor.
This is used to determine the expiration time of the FlowFile. "
+ "This attribute is not written when the FlowFile is transferred to
failure, expired or success"),
- @WritesAttribute(attribute = "wait.counter.<counterName>", description
= "If a signal exists when the processor runs, "
- + "each count value in the signal is copied.")
+ @WritesAttribute(attribute = "wait.counter.<counterName>", description
= "The name of each counter for which at least one signal "
+ + "has been present in the cache since the last time the cache was
empty "
+ + "gets copied to the current FlowFile as an attribute.")
})
@SeeAlso(classNames =
{"org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService",
"org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer",
"org.apache.nifi.processors.standard.Notify"})
@@ -110,8 +117,9 @@ public class Wait extends AbstractProcessor {
public static final PropertyDescriptor RELEASE_SIGNAL_IDENTIFIER = new
PropertyDescriptor.Builder()
.name("release-signal-id")
.displayName("Release Signal Identifier")
- .description("A value, or the results of an Attribute Expression
Language statement, which will " +
- "be evaluated against a FlowFile in order to determine the
release signal cache key")
+ .description("A value that specifies the key to a specific release
signal cache. "
+ + "To decide whether the FlowFile that is being processed by
the Wait processor should be sent to the 'success' "
+ + "or the 'wait' relationship, the processor checks the
signals in the cache specified by this key.")
.required(true)
.addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING,
true))
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
@@ -120,11 +128,12 @@ public class Wait extends AbstractProcessor {
public static final PropertyDescriptor TARGET_SIGNAL_COUNT = new
PropertyDescriptor.Builder()
.name("target-signal-count")
.displayName("Target Signal Count")
- .description("A value, or the results of an Attribute Expression
Language statement, which will " +
- "be evaluated against a FlowFile in order to determine the
target signal count. " +
- "This processor checks whether the signal count has
reached this number. " +
- "If Signal Counter Name is specified, this processor
checks a particular counter, " +
- "otherwise checks against total count in a signal.")
+ .description("The number of signals that need to be in the cache
(specified by the Release Signal Identifier) "
+ + "in order for the FlowFile processed by the Wait processor
to be sent to the ‘success’ relationship. "
+ + "If the number of signals in the cache has reached this
number, the FlowFile is routed to the "
+ + "'success' relationship and the number of signals in the
cache is decreased by this value. "
+ + "If Signal Counter Name is specified, this processor checks
a particular counter, "
+ + "otherwise checks against the total number of signals in the
cache.")
.required(true)
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
@@ -134,9 +143,9 @@ public class Wait extends AbstractProcessor {
public static final PropertyDescriptor SIGNAL_COUNTER_NAME = new
PropertyDescriptor.Builder()
.name("signal-counter-name")
.displayName("Signal Counter Name")
- .description("A value, or the results of an Attribute Expression
Language statement, which will " +
- "be evaluated against a FlowFile in order to determine the
signal counter name. " +
- "If not specified, this processor checks the total count
in a signal.")
+ .description("Within the cache (specified by the Release Signal
Identifier) the signals may belong to different counters. "
+ + "If this property is specified, the processor checks the
number of signals in the cache that belong to this particular counter. "
+ + "If not specified, the processor checks the total number of
signals in the cache.")
.required(false)
.addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.STRING,
true))
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
diff --git
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.Wait/additionalDetails.html
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.Wait/additionalDetails.html
index 4f54271..f81a2ca 100644
---
a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.Wait/additionalDetails.html
+++
b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.Wait/additionalDetails.html
@@ -15,9 +15,9 @@
limitations under the License.
-->
<head>
- <meta charset="utf-8"/>
- <title>ValidateCsv</title>
- <link rel="stylesheet" href="../../../../../css/component-usage.css"
type="text/css"/>
+ <meta charset="utf-8"/>
+ <title>Wait</title>
+ <link rel="stylesheet" href="../../../../../css/component-usage.css"
type="text/css"/>
<style>
table td:first-child {text-align: center;}
</style>
@@ -32,10 +32,10 @@ When a Wait processor is expected to process multiple
signal ids, by configuring
Also, you will need to have high level understanding on how Wait processor
works:
<ul>
- <li>Wait processor only process a single signal id at a time</li>
- <li>How frequent Wait processor runs is defined at 'Run Schedule'</li>
- <li>Which FlowFile is processed is determined by Prioritizer</li>
- <li>Not limited to Wait processor, but for all processors, the order of
queued FlowFiles in a connection is undefined if no Prioritizer is set</li>
+ <li>The Wait processor only processes a single signal id at a time</li>
+ <li>How frequently the Wait processor runs is defined in the 'Run
Schedule'</li>
+ <li>Which FlowFile is processed is determined by a Prioritizer</li>
+ <li>Not limited to the Wait processor, but for all processors, the
order of queued FlowFiles in a connection is undefined if no Prioritizer is
set</li>
</ul>
@@ -273,5 +273,101 @@ Repeat above cycle, thus F5 will not be released until
one of F1 ... F4 is relea
To mitigate such limitation, increasing 'Run Duration' is recommended. By
increasing 'Run Duration', Wait processor can keep being scheduled for that
duration. For example, with 'Run Duration' 500 ms, Wait should be able to loop
through all 5 queued FlowFiles at a single run.
</p>
+<h2>Using counters</h2>
+<p>
+ A counter is basically a label to differentiate signals within the
cache.
+ (A cache in this context is a "container" that contains signals that
have the same signal identifier.)
+</p>
+<p>
+ Let's suppose that there are the following signals in the cache
+ (note, that these are not FlowFiles on the incoming (or wait)
connection of the Wait processor, like
+ in the examples above, but release signals stored in the cache.)
+</p>
+<table>
+ <tr>
+ <th>Signal ID</th>
+ <th>Signal Counter Name</th>
+ </tr>
+ <tr>
+ <td>A</td>
+ <td>counter_1</td>
+ </tr>
+ <tr>
+ <td>A</td>
+ <td>counter_1</td>
+ </tr>
+ <tr>
+ <td>A</td>
+ <td>counter_2</td>
+ </tr>
+</table>
+<p>
+ In this state, the following FlowFile gets processed by the Wait
processor,
+ (the FlowFile has a signal_counter_name attribute and the Wait
processor is configured
+ to use the value of this attribute as the value of the Signal Counter
Name property):
+</p>
+<table>
+ <tr>
+ <th>FlowFile UUID</th>
+ <th>Signal ID</th>
+ <th>signal_counter_name</th>
+ </tr>
+ <tr>
+ <td>a-1</td>
+ <td>A</td>
+ <td>counter_3</td>
+ </tr>
+</table>
+<p>
+ Despite the fact that the cache identified by Signal ID "A" has signals
in it,
+ the FlowFile above will be sent to the 'wait' relationship, since there
is no signal
+ in the cache that belongs to the counter named "counter_3".
+</p>
+<p>
+ Let's suppose, that the state of the cache is the same as above, and
the following
+ FlowFile gets processed by the Wait processor:
+</p>
+<table>
+ <tr>
+ <th>FlowFile UUID</th>
+ <th>Signal ID</th>
+ <th>signal_counter_name</th>
+ </tr>
+ <tr>
+ <td>a-2</td>
+ <td>A</td>
+ <td>counter_1</td>
+ </tr>
+</table>
+<p>
+ The FlowFile is transmitted to the 'success' relationship, since cache
"A" has
+ signals in it and there are signals that belong to "counter_1". The
outgoing FlowFile
+ will have the following attributes and their values appended to it:
+</p>
+<ul>
+ <li>wait.counter.counter_1 : 2</li>
+ <li>wait.counter.counter_2 : 1</li>
+ <li>wait.counter.total : 3</li>
+</ul>
+<p>
+ The key point here is that counters can be used to differentiate
between signals within the
+ cache. If counters are used, a new attribute will be appended to the
FlowFile
+ passing the Wait processor for each counter. If a large number of
counters are used
+ within a cache, the FlowFile passing the Wait processor will have a
large number of
+ attributes appended to it. To avoid that, it is recommended to use
multiple caches with
+ a few counters in each, instead of one cache with many counters.
+</p>
+<p>
+ For example:
+</p>
+<ul>
+ <li>Cache identified by Release Signal ID "A" has counters: "counter_1"
and "counter_2"</li>
+ <li>Cache identified by Release Signal ID "B" has counters: "counter_3"
and "counter_4"</li>
+ <li>Cache identified by Release Signal ID "C" has counters: "counter_5"
and "counter_6"</li>
+</ul>
+<p>
+(Counter names do not need to be unique between caches, the counter name(s)
used in cache "A"
+could be reused in cache "B" and "C" as well.)
+</p>
</body>
</html>
