@Override
  public final List<? extends FileBasedSource<T>> split(
      long desiredBundleSizeBytes, PipelineOptions options) throws Exception {
    // This implementation of method split is provided to simplify subclasses. Here we
    // split a FileBasedSource based on a file pattern to FileBasedSources based on full single
    // files. For files that can be efficiently seeked, we further split FileBasedSources based on
    // those files to FileBasedSources based on sub ranges of single files.
    String fileOrPattern = fileOrPatternSpec.get();

    if (mode == Mode.FILEPATTERN) {
      long startTime = System.currentTimeMillis();
      List<Metadata> expandedFiles =
          FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata();
      List<FileBasedSource<T>> splitResults = new ArrayList<>(expandedFiles.size());
      for (Metadata metadata : expandedFiles) {
        FileBasedSource<T> split = createForSubrangeOfFile(metadata, 0, metadata.sizeBytes());
        verify(
            split.getMode() == Mode.SINGLE_FILE_OR_SUBRANGE,
            "%s.createForSubrangeOfFile must return a source in mode %s",
            split,
            Mode.SINGLE_FILE_OR_SUBRANGE);
        // The split is NOT in FILEPATTERN mode, so we can call its split without fear
        // of recursion. This will break a single file into multiple splits when the file is
        // splittable and larger than the desired bundle size.
        splitResults.addAll(split.split(desiredBundleSizeBytes, options));
      }
      LOG.info(
          "Splitting filepattern {} into bundles of size {} took {} ms "
              + "and produced {} files and {} bundles",
          fileOrPattern,
          desiredBundleSizeBytes,
          System.currentTimeMillis() - startTime,
          expandedFiles.size(),
          splitResults.size());
      return splitResults;
    } else {
      FileSystems.reportSourceLineage(getSingleFileMetadata().resourceId());
      if (isSplittable()) {
        @SuppressWarnings("unchecked")
        List<FileBasedSource<T>> splits =
            (List<FileBasedSource<T>>) super.split(desiredBundleSizeBytes, options);
        return splits;
      } else {
        LOG.debug(
            "The source for file {} is not split into sub-range based sources since "
                + "the file is not seekable",
            fileOrPattern);
        return ImmutableList.of(this);
      }
    }
  }
@Test
  public void testReadAllSplitsOfFilePattern() throws Exception {
    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);
    List<? extends BoundedSource<String>> sources = source.split(512, null);

    // Not a trivial split.
    assertTrue(sources.size() > 1);

    List<String> results = new ArrayList<>();
    for (BoundedSource<String> split : sources) {
      results.addAll(readFromSource(split, options));
    }

    List<String> expectedResults = new ArrayList<>();
    expectedResults.addAll(data1);
    expectedResults.addAll(data2);
    expectedResults.addAll(data3);

    assertThat(expectedResults, containsInAnyOrder(results.toArray()));
  }