@Rohit1997jio, I was going to answer this similar to @MattWho   who beat me to it but I will post my answer anyway in case it can help. Basically @MattWho  is correct , you can use the ExecuteScript processor to somehow simulate the retry flowfile processor and its better than RouteOnAttribute Option because when you use penalize the processor is setting idle during this time unlike the RouteOnAttribute where its always looping to unmached relationship for the period of wait time. Anyway here is my solution and in my case you dont need RouteOnAttribute but you have to add more code. I'm using groovy code for my script. The process relies on two attributes: 1- totalRetry: which incremental value to track the retry threshold every time the file is sent to retry. first time it will be set to 0. 2- isPenalized: is used to track if the file should be penalized before each retry (isPenalized == null) or its already penalized which means its ready for the next retry. The groovy script:   flowFile = session.get() if(!flowFile) return // get totalPenalized and isPenalized attributes totalRetry = flowFile.getAttribute('totalRetry') isPenalized = flowFile.getAttribute('isPenalized') // if its the first time set the value to 0 (no rety yet. first penalize) totalRetry = !totalRetry ? 0 : totalRetry.toInteger() // if the total retry has passed the threshold ( 3 in this case) then send to failure rel (expired). // Total wait time (penalize time) 3*10 sec = 30 secs if(totalRetry.toInteger()>3) { session.transfer(flowFile, REL_FAILURE) return } // if totalRetry has not passed the threshold and the file is not // penalized (isPenalized == null) then penalize and send back to upstream queue if(!isPenalized) { flowFile = session.putAttribute(flowFile, 'isPenalized', '1') flowFile = session.penalize(flowFile) session.transfer(flowFile) return } // Otherwise file has been already penalized then send to retry and increment totalRetry flowFile = session.putAttribute(flowFile, 'totalRetry', (totalRetry+1).toString()) flowFile = session.removeAttribute(flowFile, 'isPenalized') session.transfer(flowFile, REL_SUCCESS) return   You can set the Penalize period under processor SETTINGS tab   Hope this helps Thanks       ... View more

It seems that you need to process the 500 records once. In this case you dont need to start the ExecuteSQL processor nor worry about scheduling it unless you want to run it again in the future. Keep the processor stopped then  right click on it and select Run Once. This should load the total number of records one time. ... View more

Hi @Eva0 , You were close . I think the first astrik "*" was  not needed because the input json is not contained in an array brackets. Then to include the email in each data element it has to be added at the date & type level but to reference the value you need to go to 2 levels up using the @ character as in @(2,mail) where the email is located in relative to date & type. [ { "operation": "shift", "spec": { //level 2 where email originally is "dates": { // level 1 "*": { // start count here at level 0 "@(2,email)": "[&1].email", "date": "[&1].date", "type": "[&1].type" } } } } ] If you find this helpful please accept solution. Thanks ... View more