@Override
  public final BoundedReader<T> createReader(PipelineOptions options) throws IOException {
    // Validate the current source prior to creating a reader for it.
    this.validate();
    String fileOrPattern = fileOrPatternSpec.get();

    if (mode == Mode.FILEPATTERN) {
      long startTime = System.currentTimeMillis();
      List<Metadata> fileMetadata =
          FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata();
      LOG.info("Matched {} files for pattern {}", fileMetadata.size(), fileOrPattern);
      List<FileBasedReader<T>> fileReaders = new ArrayList<>();
      for (Metadata metadata : fileMetadata) {
        long endOffset = metadata.sizeBytes();
        fileReaders.add(
            createForSubrangeOfFile(metadata, 0, endOffset).createSingleFileReader(options));
      }
      LOG.debug(
          "Creating a reader for file pattern {} took {} ms",
          fileOrPattern,
          System.currentTimeMillis() - startTime);
      if (fileReaders.size() == 1) {
        return fileReaders.get(0);
      }
      return new FilePatternReader(this, fileReaders);
    } else {
      return createSingleFileReader(options);
    }
  }
@Test
  public void testCloseUnstartedFilePatternReader() throws IOException {
    PipelineOptions options = PipelineOptionsFactory.create();
    List<String> data1 = createStringDataset(3, 50);
    File file1 = createFileWithData("file1", data1);

    List<String> data2 = createStringDataset(3, 50);
    createFileWithData("file2", data2);

    List<String> data3 = createStringDataset(3, 50);
    createFileWithData("file3", data3);

    List<String> data4 = createStringDataset(3, 50);
    createFileWithData("otherfile", data4);

    TestFileBasedSource source =
        new TestFileBasedSource(new File(file1.getParent(), "file*").getPath(), 64, null);
    Reader<String> reader = source.createReader(options);
    // Closing an unstarted FilePatternReader should not throw an exception.
    try {
      reader.close();
    } catch (Exception e) {
      throw new AssertionError(
          "Closing an unstarted FilePatternReader should not throw an exception", e);
    }
  }