code: @Override
    public KeyValue<K, Iterator<V>> next() throws NoSuchElementException {
      if (combineValues) {
        skipLeftoverItems();
      }
      PeekingInputReader<KeyValue<ByteBuffer, ? extends Iterable<V>>> reader =
          lowestReaderQueue.remove();
      KeyValue<ByteBuffer, ? extends Iterable<V>> lowest = reader.next();
      ByteBuffer lowestKey = lowest.getKey();
      addReaderToQueueIfNotEmpty(reader);
      lastKey = SerializableValue.of(getByteBufferMarshaller(), lowestKey);
      if (combineValues) {
        CombiningReader values = new CombiningReader(lowestKey, lowest.getValue());
        return new KeyValue<K, Iterator<V>>(keyMarshaller.fromBytes(lowestKey.slice()), values);
      } else {
        return new KeyValue<>(keyMarshaller.fromBytes(lowestKey.slice()),
            lowest.getValue().iterator());
      }
    }
docs: Returns the next KeyValues object for the reducer. This is the entry point.
  @throws NoSuchElementException if there are no more keys in the input.