Hello Sam!  i am dealing with same issue as you, i see you made a question a year ago and i wonder, did you find a way to upload a Definition file from API rest ?  as you i have been tried and tired to build a process group with the file.json but documentation is not clear and does not offer an example to build the request.  i upload a file.json manually and checked the nifi-request.log and the only i see is this pettiotion " http://localhost:8081/process-groups/{id}/process-groups/upload" as "ok" did you find a way to build the body to send this request ?  thanks a lot, have a nice day.   ... View more

@samrathal  Apache NiFi has hardcoded return size to 100: https://github.com/apache/nifi/blob/main/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/dao/impl/StandardConnectionDAO.java#L361 I am sure this was originally done for performance and NiFi JVM heap usage reasons. The first 100 returned should be the oldest 100 in queue (keeping in mind that a connection will also show count of FlowFiles pending processing by downstream processor and count of those currently allocated to a downstream component process.  The listing only returns those pending FlowFiles and not those already owned by downstream component).What is the use case for needing to list more?  Ideally what is found in a queue should be changing rapidly, so expectation is that each listing request would be different. Listing a queue does not stop NiFi processing.  The intent is not for NiFi to ever hold FlowFiles in any connection.   So using API to  poll connection for FlowFile listings seems odd to me.  What is returned by that listing could be inaccurate milliseconds later. Also be careful with your API requests.  When a listing is performed through the browser three different request are made. 1. First listing-request is made and replicated to all nodes to get result sets.   2. Return from step 1 request gives the ID for the generated listing request being held in heap memory.  That ID is used to fetch the results in that specific listing ID 3. A DELETE request is made to remove the listing with that ID from NiFi.   *** When using API, If steps 1 and 2 are all that are being executed, the various listing request(s) will stay in heap memory. If you found any of the suggestions/solutions provided helped you with your issue, please take a moment to login and click "Accept as Solution" on one or more of them that helped. Thank you, Matt ... View more

@samrathal  Is your Windows server hosting a SFTP server that you can connect to?  If not, I'd expect GetSFTP/FetchSFTP to fail to connect.   If you can't reach your Windows file from command line on the server where NiFi resides, you are not going to be able to reach via a NiFi dataflow. 1. Setting up a windows share that you mount to your linux box would allow you to use ListFile and FetchFile processors. 2. Setting up an SFTP server on your Windows server would allow you to use ListSFTP and FetchSFTP processors. 3. Setting up SMB on your Windows server would allow you to use the ListSMB and FetchSMB processors. If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

@samrathal  1. What is the purpose of the SplitJson in your dataflow? 2. If you have 1 FlowFile with 1000 records in it, why use SplitJson to split that in to 1000 FlowFiles having 1 record each?  Why not just merge the larger FlowFiles with multiple records in it?  Or am i missing part of the use case here? --- Can you share a template of flow definition of yoru dataflow? 1. It is not clear to me how you get "X-Total-Count" and how you are adding this FlowFile attribute to every FlowFile. 2. You have configured the "Release Signal Identifier" with a boolean NiFi Expression Language (NEL) that using your example will return "false" until "fragment.count" FlowFile attribute value equals the FlowFile attribute "X-Total-Count" value.          2a. I assume you are writing "X-Total-Count" to every FlowFile coming out of the SplitJson?  How are incrementing the "fragment.count" across all FlowFile in the complete 5600 record batch.  Each FlowFile that splits into 1000 FlowFiles via splitJson will have fragment.count set to 1 - 1000.  So fragment.count would never reach 5600 unless you are handling this count somewhere else in your dataflow.         2b. If a FlowFile where value from "fragment.count" actually equals value from "X-Total-Count" attribute, your "Release Signal Identifier" will resolve to "true".  The ""Release Signal Identifier" value (true or false) in your configuration is looked up in the configured "distributed map cache server.  So where in your dataflow to you write the release signal to the distributed map cache? (usually handled by a notify processor) I am in no way implying that what you are trying to accomplish can't be done.  However, coming up with an end-to-end workable solution requires knowing all the steps in the use case along the way.   I would recommend going through the example Wait/Notify linked in my original response to get a better understanding of how wait and notify processors work together.  Then maybe you can makes some changes to your existing dataflow implementation.  With more use case details (detailed process steps) I could suggest further changes if needed. I really hope this helps you get some traction on your use case here. If you have a contract with Cloudera, you can reach out to your  account owner who could help arrange for professional services that can work with your to solution your use cases in to workable NiFi dataflows. If you found that the provided solution(s) assisted you with your query, please take a moment to login and click Accept as Solution below each response that helped. Thank you, Matt ... View more

Hi @SirV  can you please send me the code , m also stuck in the same situation.   ... View more

I am facing similar issues. Please help me out if anyone has any solution to this. ... View more

@samrathal ,   Is seems that "subtract" only works for integers and your value is a long. The longSubtract function only works if longs are passed as parameters and I don't know if there's a way to specify a long literal in Jolt (I tried 60L but that doesn't work).   The following does a little bit more work but also achieves what you want:   [ { "operation": "modify-overwrite-beta", "spec": { "currenttime": "=toLong(${now():toNumber()})", "minute": "=toLong(60)" } }, { "operation": "modify-overwrite-beta", "spec": { "timeOneMinu": "=longSubtract(@(1,currenttime), @(1,minute))" } }, { "operation": "remove", "spec": { "minute": "" } } ]     Output: { "currenttime": 1662417935173, "timeOneMinu": 1662417935113 } Cheers, André ... View more

Hi ,   where is the key,value pair are stored? Are they stored as a flowfile attribute or in the input json? there are processors like RouteOnAttribute in case they are stored as attributes where you can create your condition as dynamic property and provide the logical expression there. If the key\value pair is part of the json you can use Extract Text to Extract the key\value pair as an attribute then use RouteOnAttribute or you can try to use RouteOnContent if that will work?   If you find this helpful, please accept solution. Thanks Samer ... View more