@Fanxxx  You have what sounds like a rather complex use case here with numerous outputs, timing controls, and routing requirements.   Control Rate is very basic in nature (allow X FlowFiles to pass every X amount of time) which depending on volume of FlowFiles can lead to a backlog that ends up  resulting in most request to fail after your 5 second requirement (including new that end up delayed more the 5 seconds because they are still queued up behind other FlowFiles behind yoru ControlRate.) Cloudera offers professional Services to its licensed users that can help design and implement complex use cases.  Assisting you through the community would require considerable back and forth and exchange of information to include test files, etc. Thank you, Matt ... View more

Hi Matt,  yes it was the case of null values - we updated avro schema and set default values for enums and were able to connect and post messages to kafka. Thanks ... View more

Thanks ... View more

@Rothmny  Based on what you have shared, you should be able to accomplish this with  GenerateFlowFile and ExecuteStreamCommand processors. The GenerateFlowFile processor would be configured to generate a 0 byte FlowFile and scheduled using Cron so that it creates a FlowFile on your required schedule that will then trigger the execution of the ExecuteStreamCommand processor that will execute your python script. For calling rest-api endpoints via NiFi dataflows, you could use the InvokeHTTP processor.  Since the InvokeHTTP processor does not require and inbound connection, you could just have it as first processor in a dataflows scheduled with cron as well.  Or, if the invokeHTTP is called after your script successfully executes you could trigger with the FlowFile that exited the the ExecuteStreamCommandProcessor. Please help our community grow. If you found any of the suggestions/solutions provided helped you with solving your issue or answering your question, please take a moment to login and click "Accept as Solution" on one or more of them that helped. Thank you, Matt   ... View more

It seems after i have added necessary permission, i am able to access the context parameter via invoke HTTP   ... View more

@MarinaM Has the reply helped resolve your issue? If so, please mark the appropriate reply as the solution, as it will make it easier for others to find the answer in the future.  Thanks. ... View more

Thank you so much @MattWho for the detailed answer. The retry logic helped a lot, I have added 'RetryFlowFile' processors in between to avoid infinite loop of retry. ... View more

Thank you, @MattWho , for the additional information. I really appreciate you providing timely and clear solutions. You’re awesome! ... View more

@hus  There are two controller services you are using for your map cache: DistributedMapCacheServer - This controller service when started creates a separate map cache server on every node in a NiFi cluster. These map cache servers do not share cached entries between them. In Apache NiFi 2.x+ "Distributed" has been removed from their name to avoid confusion.  The "Max cache Entries" and "Eviction Strategy" control how cached entries are removed from the cache. DistributedMapCacheClientService - This Controller Service is used to write data to the specific Map cache server (server hostname). It also has "distributed" removed from its name as of Apache NiFi 2.x.   You are using the DetectDuplicate processor to interact with the above Controller services. While the DetectDuplicate processor has a configurable "Age Off Duration" setting,  ONLY cached entries where both the following conditions have been met will have the cache entry removed at that configured  age off: At least one duplicate has been detected. Age off duration has expired. So any cached entires for which a duplicate has not yet been detected, that entry will remain in the cache server until the "Max cache Entries" and "Eviction Strategy" settings result in the entry removal. So depending on what data you are caching, number set for "max cache Entries", and number of duplicates you detect, your cache server likely continues to grow to max and then eviction starts.  If you have a "Persistence Directory" configured, the cached data is also being written to that directory so that it is not lost in the event the NiFi instance or DistributedMapCache server is restarted.   This also means hat after a NiFi restart the persisted cache is loaded back into heap memory. Keep in mind that there are other external cache server options that do have HA, are distributed, and would not consume NiFi's heap or memory on the NiFi host if installed on a different server/host. RedisDistributedMapCacheClientService SimpleRedisDistributedMapCacheClientService HazelcastMapCacheClient CouchbaseMapCacheClient - Removed as of Apache NiFi 2.x HBase_2_ClientMapCacheService - Removed as of Apache NiFi 2.x CassandraDistributedMapCache - Removed as of Apache NiFi 2.x Please help our community grow. If you found any of the suggestions/solutions provided helped you with solving your issue or answering your question, please take a moment to login and click "Accept as Solution" on one or more of them that helped. Thank you, Matt ... View more