[
https://issues.apache.org/jira/browse/NIFI-3332?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16537865#comment-16537865
]
Koji Kawamura commented on NIFI-3332:
-------------------------------------
[~joewitt] I agree with your suggested approach, and [~doaks80]'s script
approach makes sense to address missing files caused by the assumption on
timestamps. I am going to start writing a new processor similar to Daniel's
script and using K/V store.
I come up with following spec, do you guys agree with it? If it looks
reasonable, I will create another JIRA to track the addition of new set of
WatchEntries processors. Thanks!
* Add new abstract processor AbstractWatchEntries similar to
AbstractListProcessor but uses different approach
* Target entities have: name (path), size and last-modified-timestamp
* Implementation Processors have following properties:
** 'Watch Time Range' to limit the maximum time period to hold the already
listed entries. E.g. if set as '30min', the processor keeps entities listed in
the last 30 mins.
** 'Minimum File Age' to defer listing entities potentially being written
* Any entity added but not listed ever having last-modified-timestamp older
than configured 'Watch Time Range' will not be listed. If user needs to pick
these items, they have to make 'Watch Time Range' longer. It also increases the
size of data the processor has to persist in the K/V store. Efficiency vs
reliability trade-off.
* The already-listed entities are persisted into one of supported K/V store
through DistributedMapCacheClient service. User can chose what KVS to use from
HBase, Redis, Couchbase and File (DistributedMapCacheServer with persistence
file).
* The reason to use KVS instead of ManagedState is, to avoid hammering
Zookeeper too much with frequently updating Zk node with large amount of data.
The number of already-listed entries can be huge depending on use-cases. Also,
we can compress entities with DistributedMapCacheClient as it supports putting
byte array, while ManagedState only supports Map<String, String>.
* On each onTrigger:
** Processor performs listing. Listed entries meeting any of the following
condition will be written to the 'success' output FlowFile:
*** Not exists in the already-listed entities
*** Having newer last-modified-timestamp
*** Having different size
** Already listed entries those are old enough compared to 'Watch Time Range'
are discarded from the already-listed entries.
* Initial supporting target is Local file system, FTP and SFTP
> Bug in ListXXX causes matching timestamps to be ignored on later runs
> ---------------------------------------------------------------------
>
> Key: NIFI-3332
> URL: https://issues.apache.org/jira/browse/NIFI-3332
> Project: Apache NiFi
> Issue Type: Bug
> Components: Core Framework
> Affects Versions: 0.7.1, 1.1.1
> Reporter: Joe Skora
> Assignee: Koji Kawamura
> Priority: Critical
> Fix For: 1.4.0
>
> Attachments: Test-showing-ListFile-timestamp-bug.log,
> Test-showing-ListFile-timestamp-bug.patch, listfiles.png
>
>
> The new state implementation for the ListXXX processors based on
> AbstractListProcessor creates a race conditions when processor runs occur
> while a batch of files is being written with the same timestamp.
> The changes to state management dropped tracking of the files processed for a
> given timestamp. Without the record of files processed, the remainder of the
> batch is ignored on the next processor run since their timestamp is not
> greater than the one timestamp stored in processor state. With the file
> tracking it was possible to process files that matched the timestamp exactly
> and exclude the previously processed files.
> A basic time goes as follows.
> T0 - system creates or receives batch of files with Tx timestamp where Tx
> is more than the current timestamp in processor state.
> T1 - system writes 1st half of Tx batch to the ListFile source directory.
> T2 - ListFile runs picking up 1st half of Tx batch and stores Tx timestamp
> in processor state.
> T3 - system writes 2nd half of Tx batch to ListFile source directory.
> T4 - ListFile runs ignoring any files with T <= Tx, eliminating 2nd half Tx
> timestamp batch.
> I've attached a patch[1] for TestListFile.java that adds an instrumented unit
> test demonstrates the problem and a log[2] of the output from one such run.
> The test writes 3 files each in two batches with processor runs after each
> batch. Batch 2 writes files with timestamps older than, equal to, and newer
> than the timestamp stored when batch 1 was processed, but only the newer file
> is picked up. The older file is correctly ignored but file with the matchin
> timestamp file should have been processed.
> [1] Test-showing-ListFile-timestamp-bug.patch
> [2] Test-showing-ListFile-timestamp-bug.log
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
