code: @Override
  	public void processEvent(QueryBatch batch) {
  		// resultsSoFar is an AtomicLong so it can be safely updated across many threads
  		final long jobResultsSoFar = batch.getJobResultsSoFar();
  		final long newResultsSoFar = this.resultsSoFar.updateAndGet(operand ->
  			jobResultsSoFar > operand ? jobResultsSoFar : operand
  		);
  		boolean resultsSoFarWasUpdated = jobResultsSoFar == newResultsSoFar;
  
  		if (resultsSoFarWasUpdated && consumers != null) {
  			double timeSoFar = ((double) System.currentTimeMillis() - startTime) / 1000;
  
  			/**
  			 * The initial totalResults may have been incorrectly set to a value lower than jobResultsSoFar; if this
  			 * occurs, use jobResultsSoFar as the value passed to the ProgressUpdate object. totalResults is not
  			 * updated though in case there's a need to know what the initial value was.
  			 */
  			long totalForThisUpdate = jobResultsSoFar > this.totalResults && this.totalResults > 0 ? jobResultsSoFar : this.totalResults;
  
  			ProgressUpdate progressUpdate = newProgressUpdate(batch, startTime, totalForThisUpdate, timeSoFar);
  
  			for (Consumer<ProgressUpdate> consumer : consumers) {
  				invokeConsumer(consumer, progressUpdate);
  			}
  		}
  	}
docs: Batches arrive in random order, so a ProgressUpdate is created and sent to each Consumer only if the
  value of "getJobResultsSoFar" on the QueryBatch exceeds the number of results seen so far.
  <p>
  For example, if there are 2 batches, and batch 2 is processed first by this listener followed by batch 1, a
  ProgressUpdate is only created when batch 2 is processed.
  
  @param batch