code: @Override
    public List<? extends InputReader<KeyValue<ByteBuffer, Iterator<ByteBuffer>>>> createReaders() {
      Marshaller<ByteBuffer> byteBufferMarshaller = Marshallers.getByteBufferMarshaller();
      Marshaller<KeyValue<ByteBuffer, ? extends Iterable<ByteBuffer>>> marshaller =
          Marshallers.getKeyValuesMarshaller(byteBufferMarshaller, byteBufferMarshaller);
      ImmutableList.Builder<InputReader<KeyValue<ByteBuffer, Iterator<ByteBuffer>>>> result =
          ImmutableList.builder();
      for (int shard = 0; shard < filesByShard.getShardCount(); shard++) {
        List<InputReader<KeyValue<ByteBuffer, Iterator<ByteBuffer>>>> readers = new ArrayList<>();
        for (List<String> group : Lists.partition(filesByShard.getFilesForShard(shard).getFileNames(),
            mergeFanin)) {
          GoogleCloudStorageFileSet fileSet =
              new GoogleCloudStorageFileSet(filesByShard.getBucket(), group);
          readers.add(createReaderForShard(marshaller, fileSet));
        }
        result.add(new ConcatenatingInputReader<>(readers));
      }
      return result.build();
    }
docs: Creates multiple merging readers for each shard using {@link #createReaderForShard} below.
  These are combined into a single reader via concatenation. The resulting input stream will be
  in order except when the input switches from one merging reader to the next the key will very
  likely go backwards.