code: void doStream(DataStreamerProcessor<E> processor,
              List<DataStreamingEventIterator<E>> itrs) throws
              DataSourceReadException {
          assert processor != null : "processor cannot be null";
          assert itrs != null : "itrs cannot be null";
          if (this.itr == null) {
              this.itr = new DataStreamerIterator(itrs);
          }
          boolean stopOnNext = false;
          while (this.itr.hasNext() && !stopOnNext) {
              String keyId = processor.getKeyId();
              String nextKeyId = this.itr.getNextKeyId();
              if (keyId != null && !keyId.equals(nextKeyId)) {
                  stopOnNext = true;
              }
              processor.execute(this.itr.next());
          }
  
          if (!stopOnNext) {
              processor.finish();
          }
      }
docs: Groups the data in a list of {@link DataStreamingEventIterator}s by
  key id.
  
  @param processor a {@link DataStreamerProcessor} that helps with the
  grouping.
  @param itrs the data to be grouped.
  @throws DataSourceReadException if an error occurred when iterating over
  the {@link DataStreamingEventIterator}s.