@Rohit1997jio, As @joseomjr already pointed out, by doing this, you defeat the single purpose of using Kafka. As you already know, Kafka is a stream-processing platform and basically function in a very basic way as a queue of messages. Integrating Kafka with NiFi, especially using the processor ConsumeKafka, you basically create a bucket at the end of the queue. As long as messages are present in the queue (Kafka in your case), you will have messages arriving in your bucket (you NiFi processing layer in your case). When you do not have any messages in the kafka system, your ConsumeKafka processor will be in a let's call it idle state, meaning that it will not waste resources in vain - it will however use some resources to check whether new messages arrived or not. That being said, I see no point in trying to kill a connection which is NOT affecting the involved systems in any way and basically defeats the entire purpose of using NiFi and Kafka. However, if you still want to achieve this, you will need to put some extra effort in doing this. First of all, you need to create a flow which checks the state of the desired processor using the NiFi's REST API - achievable in many ways, like InvokeHTTP, ExecuteStreamCommand, etc. If nothing has been done in the past 5 minutes (displayed in the JSON received as response from REST API) you will activate an InvokeHTTP in which you call again the REST API for STOPPING the ConsumeKafka Processor. ... View more

@Chaitanya_, First of all, why the infinite loop? What is the point of that and what where you actually trying to achieve? ( just out of curiosity) Now, in terms of your problem, I can give you two possible scenarios which you could try, which might eventually help you: 1) Add a new processor on your Canvas (LogMessage for example) and try moving the queue from the center funnel towards the right funnel into the new processor. Please make sure that all your processors are stopped and disabled and your queues are empty. This should allow you to move the queue without any issues, while avoiding the infinite loop, which will eventually help you to remove the funnels from your canvas. Another thing, try also removing the purple highlighted queue, so you are certain that the loop is no longer a loop. Afterwards, you should be able to remove all the queues, starting from right to left.   2) This might be a little hard and requires plenty of attention (and it is not really recommended), but in times of desperation, you could try manually modifying the flow.xml.gz and the flow.json.gz and remove the parts of those funnels. You then can upload the new version of files in all NiFi nodes and you should no longer see those funnels on your canvas. However, before doing this, make sure that you create a backup of those files, in case you mess something up. Nevertheless, this is not really recommended so I highly advise you to try with the first solution before even trying this one. PS: make sure that all your nodes are up and running. Or stop the nodes, work on a single node and copy the flow.xml and flow.json to all the other nodes and start them. Hope it helps! ... View more

@user001, How I would do this: - Create an UpdateRecord Processor, where I define a JsonTreeReader, for reading the input file and a JsonRecordSetWriter, for writing the newly formatted file. - Within this same processor, I would add a new property where I define the path to the column you are trying to modify. In your case, assuming that this is the entire JSON File, you would have /Key/value[*]/Date. - The value for this newly generated property should be the transformation you are trying to achieve. In your case, it would be ${field.value:toDate("MM/yy"):format("yyyy-MM")} - Next, within the same exact processor, I would modify the Replacement Value Strategy from the default value into Literal Value. And that's pretty much all you have to do. ... View more