code: private void skipLeftoverItems() {
      if (lastKey == null || lowestReaderQueue.isEmpty()) {
        return;
      }
      boolean itemSkiped;
      do {
        PeekingInputReader<KeyValue<ByteBuffer, ? extends Iterable<V>>> reader =
            lowestReaderQueue.remove();
        itemSkiped = skipItemsOnReader(reader);
        addReaderToQueueIfNotEmpty(reader);
      } while (itemSkiped);
    }
docs: It is possible that a reducer does not iterate over all of the items given to it for a given
  key. However on the next callback they expect to receive items for the following key. this
  method skips over all the left over items from the previous key they did not read.