We have seven different branches/lines of flow and at the terminal processors, we want to merge the output FlowFiles. These seven branches starts from a singe GenerateFlowFile, which in turn is triggered to run 3 times a day (1 AM, 7 AM and 12 PM). However, the execution time of the branches are different, so we need for each one of them to finish. At the merging part (funnel + MergeContent processor), we use the batch-time (trigger timestamp) as an attribute in order to bundle the 7 output FlowFiles. Things work out well for the most part, until one time, one of the branches has some mistake that it outputted two FlowFiles for same cycle instead of one. So for example, these FlowFiles were bundled together, while Branch 7 is not yet done: FlowFile1 FlowFile2 FlowFile3 FlowFile4 FlowFile5 FlowFile6 FlowFile6 Assuming we use a FlowFile attribute identifying the branch name, how can we bundle the same batch-time FlowFiles while guaranteeing that all branches are covered in the bundle? ... View more

I have an InvokeHTTP processor and the REST endpoint it calls requires PATCH method. However, when we send the FlowFile (we get the authentication part OK), we are getting: method PATCH must have a request body For the InvokeHTTP processor, I'm pretty sure that the we configured the Send Message Body to true (which is the default) and Content-Type as application/json. We enabled the endpoint to accept POST request, for the sake of testing and we've got the JSON payload accepted. So I wonder if there is an issue if we use PATCH. ... View more

Is it possible to place a NiFi processors name or id to a flowfile that it outputs? For example a Processor InvokeHTTP I named as CallRESTEndpoint, the flowfile that would come out of it will have attribute, like origin:CallRESTEndpoint? Or if possible to create a chain of origin that will append the processor's name or id to this attribute. I need this in error / failure handling. If you can share a better idea in tracing errors / failure, that'd be a good idea. (But please answer the original question also, if there is one.) ... View more

We have a custom query that we wish to put inside the ExecuteSQL processor, but this one proves a bit tricky: SELECT * FROM mytable AS OF TIMESTAMP TO_TIMESTAMP('${now():format('yyyy-MM-dd HH:mm:ss')}', 'YYYY-MM-DD HH24:MI:SS'); The processor is CRON driven where we run it every 15 minutes. The expression language inside the first single quotes is NiFi's expression language to get the current time in the format specified. Note that the expression language: ${now():format('yyyy-MM-dd HH:mm:ss')} itself uses single quotes. I tried substituting double quotes for the Oracle query, but Oracle won't accept the syntax as valid. So how do we substitute the intended date-time string inside the first single-quotes? ... View more

We have an SFDC REST endpoint (let's call this UpdateSFDC) where we stream and post database changes, in JSON format, to reflect it to SalesForce object store. Before we get to that endpoint, we need to Login to SalesForce first to this endpoint: https://test.salesforce.com/services/oauth2/token So we use InvokeHTTP (let's call this Login2SFDC) and appends the parameters to the Remote URL field in the processor: https://test.salesforce.com/services/oauth2/token?grant_type=password&client_id=<client_id>&client_secret=<client_secret>&username=<username>&password=<password + key> This will respond with a token which we need to extract and add as an attribute in every JSON format flowfile (let's call this CreateSignedRequest). Here's a NiFi schematic: The Login2SFDC is configure this way: However, this is quite inefficient because for every JSON flowfile, we need to login to SalesForce everytime, which will add overhead, not to mention the risk that Login2SFDC REST endpoint will eventually throw some sort of a max limit request reached. Rather, we want to login once and as long as the token is valid (not yet expired), then we will just add the token as an attribute to the JSON flowfile and go to UpdateSFDC REST endpoint straight (via CreateSignedRequest). So I will convert the Login2SFDC processor from InvokeHTTP to ExecuteScript. I have 3 questions for this problem: 1. How do I this in ECMAScript and/or Python? When I use their respective native HTTP client functionalities, they do not seem to work. Should I be importing and using Java libraries (like HttpClient) instead in which case, the scripting language is simply a wrapper to Java libraries? 2. I am successful in getting a response from Login2SFDC REST endpoint when in NiFi. I tried writing a corresponding Java code (using same exact endpoint and credentials) using HttpClient library: CloseableHttpClient client = HttpClients.createDefault(); String urlStr = "https://test.salesforce.com/services/oauth2/token?grant_type=password&client_id=<client_id>&client_secret=<client_secret>&username=<username>&password=<password + key>"; HttpPost httpPost = new HttpPost(urlStr); String jsonPayload = "{\"qty\":100,\"name\":\"iPad 4\"}"; StringEntity entity = new StringEntity(jsonPayload); httpPost.setEntity(entity); httpPost.setHeader("Accept", "application/json"); httpPost.setHeader("Content-type", "application/json"); CloseableHttpResponse response = client.execute(httpPost); System.out.println(response.getStatusLine().getStatusCode()); InputStream is = response.getEntity().getContent(); String responseStr = IOUtils.toString(is, "UTF-8"); System.out.println(responseStr); client.close(); but I get this response: Status Code 400 {"error":"invalid_grant","error_description":"authentication failure"} Am I missing something in the Java code? 3. Is there a better NiFi flow design for this kind of problem? Hi @Matt Burgess, can you help? ... View more