I have to do some custom preprocessing tasks on a huge data file (~200GB). currently, its works as below way. select * from table preprocessing line by line return a new single flow file so I decided to convert the above approach to the below way. get the row count from the user (let's assume the user gives 1000) execute select * query as resultSet read the results line by line (rs.next()) when the line count reaches 1000 return the flow file and continues to other lines So my approach is as below onTrigger public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { logger = getLogger(); FlowFile flowFile = session.get(); if (flowFile == null) { return; } try { final Long rowLimit = context.getProperty(ProcessorUtils.MAX_RECORD).evaluateAttributeExpressions(flowFile).asLong(); Connection conn = DriverManager.getConnection( // db connection properties ); Statement stm = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); ResultSet rs = stm.executeQuery("sql query"); Map<String, String> flowFileAttributes = flowFile.getAttributes(); process( rs, session, flowFileAttributes, rowLimit, ); FlowFile stateFlowFile = session.create(); session.putAttribute(stateFlowFile, "processing_status", "end"); session.putAttribute(stateFlowFile, "record_count", "0"); session.transfer(stateFlowFile, GPReaderProcessorUtils.STATUS); // working line } catch (Exception e) { logger.warn(" conn " + e); session.transfer(flowFile, GPReaderProcessorUtils.FAILURE); } } Recursion Approach for termination based on line count private void process(ResultSet rs, ProcessSession session, Map<String, String> flowFileAttributes, Long rowLimit) throws SQLException { try{ logger.info("-> start processing with row limit = " + rowLimit); AtomicInteger mainI = new AtomicInteger(0); FlowFile flowFile = session.write(session.putAllAttributes(session.create(), flowFileAttributes), (OutputStream out) -> { int i = 0; Map<String, String> preProcessResults = null; try { String res = ""; while (i < rowLimit && rs.next()) { //preprocessing happens here i++; mainI.set(i); out.write(preprocess results.toString().getBytes(StandardCharsets.UTF_8)); } }catch (SQLException e) { e.printStackTrace(); } } logger.info("gp-log ->"+ (String.valueOf(i))); out.close(); }); FlowFile stateFlowFile = session.create(); session.putAttribute(stateFlowFile, "processing_status", "processing"); session.putAttribute(stateFlowFile, "record_count", mainI.toString()); session.transfer(stateFlowFile, GPReaderProcessorUtils.STATUS); // state relationship session.transfer(flowFile, GPReaderProcessorUtils.SUCCESS); // preprocessed flow files returns if(!rs.isAfterLast() && mainI != 0 && !rs.isLast()){ // recurrsion call logger.info("gp-log -> recursion call" ); process(rs, session,flowFileAttributes,column,rowLimit); } }catch (Exception e){ logger.info(e.getMessage()); logger.error(e.getMessage()); session.transfer(session.putAllAttributes(session.create(),flowFileAttributes), GPReaderProcessorUtils.FAILURE); } } Expected Behaviour -> while processing this one return completed rows as flow files Current Behaviour -> after finishing all return all flow files (generated in recursion) once. please advise on this. ... View more

we have set up 4 node cluster and all other 3 nodes start without any exceptions. and in NIFI ui it shows our down node is in connecting state. but our down NIFI node is not running. its logs as below. 2022-10-17 19:04:51,860 INFO [main] org.apache.nifi.io.socket.SocketListener Now listening for connections from nodes on port 11443 2022-10-17 19:04:51,950 WARN [main] o.a.n.c.l.e.CuratorLeaderElectionManager Failed to close Leader Selector for Cluster Coordinator java.lang.IllegalStateException: Already closed or has not been started at com.google.common.base.Preconditions.checkState(Preconditions.java:173) at org.apache.curator.framework.recipes.leader.LeaderSelector.close(LeaderSelector.java:270) at org.apache.nifi.controller.leader.election.CuratorLeaderElectionManager.stop(CuratorLeaderElectionManager.java:165) at org.apache.nifi.controller.FlowController.shutdown(FlowController.java:1120) at org.apache.nifi.controller.StandardFlowService.stop(StandardFlowService.java:349) at org.apache.nifi.web.server.JettyServer.start(JettyServer.java:1016) at org.apache.nifi.NiFi.<init>(NiFi.java:158) at org.apache.nifi.NiFi.<init>(NiFi.java:72) at org.apache.nifi.NiFi.main(NiFi.java:297) java.lang.InterruptedException: sleep interrupted at java.lang.Thread.sleep(Native Method) at org.apache.nifi.controller.queue.clustered.client.async.nio.NioAsyncLoadBalanceClientTask.run(NioAsyncLoadBalanceClientTask.java:91) at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) at java.util.concurrent.FutureTask.run(FutureTask.java:266) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) 2022-10-17 19:04:51,957 ERROR [Load-Balanced Client Thread-58] o.a.n.c.q.c.c.a.n.NioAsyncLoadBalanceClientTask Failed to communicate with peer while trying to load balance data across the cluster java.lang.InterruptedException: sleep interrupted 2022-10-17 19:04:52,495 INFO [main] o.apache.nifi.controller.FlowController Controller has been terminated successfully. 2022-10-17 19:04:52,501 INFO [main] org.apache.nifi.io.socket.SocketListener Socket Listener has been terminated successfully. 2022-10-17 19:04:52,502 WARN [Cluster Socket Listener] org.apache.nifi.io.socket.SocketListener Failed to communicate with Unknown Host due to java.net.SocketException: Socket closed java.net.SocketException: Socket closed at java.net.PlainSocketImpl.socketAccept(Native Method) at java.net.AbstractPlainSocketImpl.accept(AbstractPlainSocketImpl.java:409) at java.net.ServerSocket.implAccept(ServerSocket.java:545) at sun.security.ssl.SSLServerSocketImpl.accept(SSLServerSocketImpl.java:348) at org.apache.nifi.io.socket.SocketListener$2.run(SocketListener.java:112) at java.lang.Thread.run(Thread.java:748) 2022-10-17 19:04:52,502 ERROR [main] org.apache.nifi.web.server.JettyServer Unable to load flow due to: java.util.zip.ZipException: invalid stored block lengths java.util.zip.ZipException: invalid stored block lengths at java.util.zip.InflaterInputStream.read(InflaterInputStream.java:164) at java.util.zip.GZIPInputStream.read(GZIPInputStream.java:117) at java.io.FilterInputStream.read(FilterInputStream.java:107) at org.apache.nifi.util.file.FileUtils.copy(FileUtils.java:303) at org.apache.nifi.persistence.StandardXMLFlowConfigurationDAO.load(StandardXMLFlowConfigurationDAO.java:110) at org.apache.nifi.controller.StandardFlowService.createDataFlow(StandardFlowService.java:608) at org.apache.nifi.controller.StandardFlowService.load(StandardFlowService.java:467) at org.apache.nifi.web.server.JettyServer.start(JettyServer.java:1009) at org.apache.nifi.NiFi.<init>(NiFi.java:158) at org.apache.nifi.NiFi.<init>(NiFi.java:72) at org.apache.nifi.NiFi.main(NiFi.java:297) 2022-10-17 19:04:52,503 WARN [main] org.apache.nifi.web.server.JettyServer Failed to start web server... shutting down. java.lang.Exception: Unable to load flow due to: java.util.zip.ZipException: invalid stored block lengths at org.apache.nifi.web.server.JettyServer.start(JettyServer.java:1019) at org.apache.nifi.NiFi.<init>(NiFi.java:158) at org.apache.nifi.NiFi.<init>(NiFi.java:72) at org.apache.nifi.NiFi.main(NiFi.java:297) 2022-10-17 19:04:52,504 INFO [Thread-1] org.apache.nifi.NiFi Initiating shutdown of Jetty web server... 2022-10-17 19:04:52,524 INFO [Thread-1] o.eclipse.jetty.server.AbstractConnector Stopped ServerConnector@4e3931{SSL,[ssl, http/1.1]}{bdadataf15.ncell.com.np:9443} 2022-10-17 19:04:52,525 INFO [Thread-1] org.eclipse.jetty.server.session node0 Stopped scavenging we didn't do any changes in properties and we deployed one custom processor since its threw exceptions while processing we revert it back also. ... View more