title: how to transfer flowfiles one by one using a custom processor in NIFI
body: <p>I'm programming a custom processor in Nifi v 1.3</p>
  
  <p>The processor executes an SQL query read from the resultset and transforms every row to json document and stores it into an ArrayList, finally it transfers every 1000 document (fetchSize param) to a flowfile, this work for me, but it sends all flowFiles at once.</p>
  
  <p>What I want is for it to transfer each flowfile independently when I call transferFlowFile method without waiting for the end of the onTrigger method to transfer everything at once.</p>
  
  <p>here the code :</p>
  
  <pre><code>public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
      FlowFile fileToProcess = null;
      if (context.hasIncomingConnection()) {
         fileToProcess = session.get();
         if (fileToProcess == null &amp;&amp; context.hasNonLoopConnection()) {
            return;
         }
      }
  
      final ResultSet resultSet = st.executeQuery();
      final ResultSetMetaData meta = resultSet.getMetaData();
      final int nrOfColumns = meta.getColumnCount();
      List&lt;Map&lt;String, Object&gt;&gt; documentList = new ArrayList&lt;&gt;();
  
          while (resultSet.next()) {
  
             final AtomicLong nrOfRows = new AtomicLong(0L);
             cpt++;
  
             Map&lt;String, Object&gt; item = new HashMap&lt;&gt;();
             for (int i = 1; i &lt;= nrOfColumns; i++) {
                 int javaSqlType = meta.getColumnType(i);
                 String nameOrLabel = StringUtils.isNotEmpty(meta.getColumnLabel(i)) ? meta.getColumnLabel(i)
                      : meta.getColumnName(i);
                 Object value = null;
                 value = resultSet.getObject(i);
                 if (value != null) {
                    item.put(nameOrLabel, value.toString());
                 }
             }
             Document document = new Document(item);
             documentList.add(document);
  
             if (fetchSize!=0 &amp;&amp; cpt % fetchSize == 0) {
              FlowFile flowFile = session.create();
              transferFlowFile(flowFile, session, documentList, fileToProcess, nrOfRows, stopWatch);
             }
         }
  
         if (!documentList.isEmpty()) {
            final AtomicLong nrOfRows = new AtomicLong(0L);
            FlowFile flowFile = session.create();
            transferFlowFile(flowFile, session, documentList, fileToProcess, nrOfRows, stopWatch);
         }
  }
  
  
  public void transferFlowFile(FlowFile flowFile, ProcessSession session, List&lt;Map&lt;String, Object&gt;&gt; documentList,
          FlowFile fileToProcess, AtomicLong nrOfRows, StopWatch stopWatch) {
  
      flowFile = session.write(flowFile, out -&gt; {
          ObjectMapper mapper = new ObjectMapper();
          IOUtils.write(mapper.writeValueAsBytes(documentList), out);
      });
  
      documentList.clear();
  
      flowFile = session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
  
      session.getProvenanceReporter().modifyContent(flowFile, "Retrieved " + nrOfRows.get() + " rows",
              stopWatch.getElapsed(TimeUnit.MILLISECONDS));
  
      session.transfer(flowFile, REL_SUCCESS);
  }
  </code></pre>