code: private void scheduleRowProcessing(RowProcessingPublishers publishers, LifeCycleHelper lifeCycleHelper,
              JobCompletionTaskListener jobCompletionTaskListener, AnalysisJobMetrics analysisJobMetrics) {
  
          logger.info("Created {} row processor publishers", publishers.size());
  
          final List<TaskRunnable> finalTasks = new ArrayList<TaskRunnable>(2);
          finalTasks.add(new TaskRunnable(null, jobCompletionTaskListener));
          finalTasks.add(new TaskRunnable(null, new CloseReferenceDataTaskListener(lifeCycleHelper)));
  
          final ForkTaskListener finalTaskListener = new ForkTaskListener("All row consumers finished", _taskRunner,
                  finalTasks);
  
          final TaskListener rowProcessorPublishersDoneCompletionListener = new JoinTaskListener(publishers.size(),
                  finalTaskListener);
  
          final Collection<RowProcessingPublisher> rowProcessingPublishers = publishers.getRowProcessingPublishers();
          for (RowProcessingPublisher rowProcessingPublisher : rowProcessingPublishers) {
              logger.debug("Scheduling row processing publisher: {}", rowProcessingPublisher);
              rowProcessingPublisher.runRowProcessing(_resultQueue, rowProcessorPublishersDoneCompletionListener);
          }
      }
docs: Starts row processing job flows.
  
  @param publishers
  @param analysisJobMetrics
  
  @param injectionManager