@rki    Your NiFi Expression Language (EL) statement shared expects that the inbound FlowFile already has FlowFile attribute named "end_time" with some value assigned to it.   What does that value look like?  How was it created? ${end_time:lt(${now():toNumber():minus(86400000)})}   Let's break down the embedded NiFi EL statement first: ${now():toNumber():minus(86400000)} The now() function returns the current timestamp. The toNumber() function converts that timestamp in to milliseconds since midnight GMT Jan 1, 1970. The minus() function subtracts the number passed to the function (86400000) from the above calculated milliseconds.   Assuming that the "end_time" attribute returns some number that represents the number of milliseconds since midnight GMT Jan 1, 1970 also and that number is less than the value calculated by the embedded NiFi EL, the NiFi EL will return "true".  Essential all files were the end_time is older then 24 hours from the current timestamp.   The FlowFile would then get routed to the relationship named by your RouteOnAttribute custom property. If false is returned and no other custom properties match, the FlowFile would be routed to unmatched.   If you are really trying to only route FlowFiles were the "end_time" milliseconds falls within the last 24 hours only, then you would want to use the ge() function instead of the lt() function.   Hope this helps, Matt ... View more

@sunilb    You may want to look at using the listS3 processor to list the files from your S3 bucket.  This will produce one 0 byte (actual file content is not retrieved by this processor) FlowFile for each S3 file that is listed.   Each of these generated FlowFile will have attributes/metadata about the file that was listed.  This includes the "filename".   You can then route the success relationship from the listS3 processor to a RouteOnAttribute processor where you route those FlowFiles where the "filename" attribute value ends with ".txt" on to a FetchS3Object processor (This processor uses the "filename" attribute from the inbound FlowFile to fetch the actual content for that S3 file and add it to the FlowFile).   Any FlowFile where the filename attribute does not end in ".txt" could just be auto-terminated. RouteOnAttribute configuration:     Here is an example of what this portion of the dataflow would look like:   The connection between RouteOnAttribute and FetchS3Object processors should be configured to use the Round Robin Load Balancing Strategy if your NiFi is setup as a cluster.  The ListS3  processor should only be configured to run on the NiFi cluster's primary node (you'll notice the mall "P" on the icon of the listS3 processor in upper left corner).  So the load balancing strategy will redistribute the listed FlowFiles amongst all nodes in your cluster before actually fetching the content for more efficient/performant use of resources.   Hope this helps, Matt   ... View more

@apocolis    I recommend adding the ValidateRecord before your PublishKafkaRecord processor to filter out the invalid records from your dataflow. Hope this helps, Matt ... View more

@GKrishan  You want to be very careful when setting safety valve in CM to override existing default property values in NiFi files. @wengelbrecht screenshot shows creating a safety valve that would override "java.arg.2" in the NiFi bootstrap.conf file with a new value of "Xmx1024m".  The problem here is that "java.arg.2" is used to set Xms while "java.arg.3" is used to set Xmx.  So you would end up with two properties defining Xmx and missing the property for Xms.   Here is an example setting override safety valves in CM for both Xms and Xmx: Hope this helps, Matt ... View more

@hhibigdata    I am not clear on your setup based on your comments:  I have to configure one more NiFi Cluster(Standalone)? My NiFi Cluster is 4(3 clustering, 1 standalone)?   ---  The List based processor are not cluster friendly because the non NiFi protocols they are built for are not cluster friendly.  All this means is that these processors must be configured in your NiFi cluster with an "Execution" of "Primary Node" so that it will only ever be running on one node at a time.  You should not have two different NiFi installs.  In between the ListSFTP and FetchSFTP processors you should be redistributing the listed files via the load balanced strategy options on the connection. --- NiFi clusters require zookeeper and zookeeper requires quorum meaning you should have an odd number of ZK nodes (3 or 5 recommended).  This same ZK will also be used to store cluster state for thes non cluster friendly processors, so that when a primary node changes nodes, the new node will pull last known state from ZK so that the list based processors continue to list from where previously elected primary node left off. So two things i suggest you check on: 1. That the zookeeper "Connect String" is correct in your state-management.xml "zk-provider". It should be a comma separated list of 3 to 5 ZK <hostname>:<port> 2. That the "nifi.zookeeper.connect.string=" has been properly setup in the nifi.properties file. It should be a comma separated list of 3 to 5 ZK <hostname>:<port> *** Generally both use the same ZK connect string and same ZK root node.   Hope this helps, Matt   ... View more

@emanueol    The intent of templates was to allow users to create re-useable dataflows or distribute dataflows to another NiFi installation. Since you can not have more than one component with the same uuid, the uuids are randomized when both creating the template and each time the components from a template are instantiated to the canvas.   The best way to identify your component in the xml is by uniquely naming your components.  By default a components name will be the same as the component type; however, users can modify the name to whatever they like.   Hope this helps, Matt ... View more

@frassis    The error message you have encountered indicates and issue with the certificates you are using to secure your NiFi nodes.   javax.net.ssl.SSLPeerUnverifiedException: Hostname <my_fqdn_is_here> not verified: certificate: sha256/716mOuXyoAKqzNrXrNnG2ozHXwN8WWJsVxzWzfQzpNV= DN: CN=xxx-xxxx-xxxx.xxx.xxx.net, OU=XXXXXXXXX XXXXXX, O=XXXXX, L=XXXXXX, ST=XXXXXX XXXXX, C=CA subjectAltNames: [] Jetty no longer uses the DN to verify hostnames and now requires that the certificates include at least 1 Subject Alternative Name (SAN) entry that matches the hostname of the server on which it is being used. As you can see from the ERROR output, it indicates you have no SAN entries in your cert.   subjectAltNames: []   You will need to generate new certificates and keystores for your NiFi nodes. When doing so keep in mind the following: 1. Keystore may contain ONLY 1 PrivateKeyEntry 2. The PrivateKeyEntry MUST support both "clientAuth" and "serverAuth" 3. The PrivateKeyEntry MUST contain at least 1 SAN entry matching the hostname of the server where keystore will be used. 4. The Keystore and Key passwords must be the same. Or no key password set. Thank you, Matt ... View more

@littlesea374    The WAIT processor requires a release signal that is typically created using the NOTIFY processor. So that really will not help here. Perhaps you could try setting a penalty on each FlowFile.  Penalized FlowFiles are not processed by the follow on processor until the penalty duration has ended.  This can be done using the ExecuteScript processor after listSFTP:   You then set length of "Penalty Duration" via the settings tab.  Set penalty high enough to ensure the file writes have completed.  Of course this does introduce some latency. What this will not help with is listSFTP still listing the same files multiple times.  As data is written, the timestamp written on that source FlowFile updates, which means it will get listed again as if it is a new file. but the delay here allows full data to be written and then perhaps you can use a detectDuplicate processor to remove duplicates based on filename before you actually fetch the content. just some thoughts here, but that JIra is probably best path....   Matt ... View more