code: private MergingReader<K, V> createReaderForShard(
        Marshaller<KeyValue<ByteBuffer, ? extends Iterable<V>>> marshaller,
        GoogleCloudStorageFileSet reducerInputFileSet) {
      ArrayList<PeekingInputReader<KeyValue<ByteBuffer, ? extends Iterable<V>>>> inputFiles =
          new ArrayList<>();
      GoogleCloudStorageLevelDbInput reducerInput =
          new GoogleCloudStorageLevelDbInput(reducerInputFileSet, DEFAULT_IO_BUFFER_SIZE);
      for (InputReader<ByteBuffer> in : reducerInput.createReaders()) {
        inputFiles.add(new PeekingInputReader<>(in, marshaller));
      }
      return new MergingReader<>(inputFiles, keyMarshaller, true);
    }
docs: Create a {@link MergingReader} that combines all the input files the reducer to provide a
  global sort over all data for the shard.
  
  (There are multiple input files in the event that the data didn't fit into the sorter's
  memory)
  
  A {@link MergingReader} is used to combine contents while maintaining key-order. This requires
  a {@link PeekingInputReader}s to preview the next item of input.
  
  @returns a reader producing key-sorted input for a shard.