@Override
  public final long getEstimatedSizeBytes(PipelineOptions options) throws IOException {
    // This implementation of method getEstimatedSizeBytes is provided to simplify subclasses. Here
    // we perform the size estimation of files and file patterns using the interface provided by
    // FileSystem.
    String fileOrPattern = fileOrPatternSpec.get();

    if (mode == Mode.FILEPATTERN) {
      Long maybeNumBytes = filesSizeBytes.get();
      if (maybeNumBytes != null) {
        return maybeNumBytes;
      }

      long totalSize = 0;
      List<Metadata> allMatches = FileSystems.match(fileOrPattern, emptyMatchTreatment).metadata();
      for (Metadata metadata : allMatches) {
        totalSize += metadata.sizeBytes();
      }
      LOG.info(
          "Filepattern {} matched {} files with total size {}",
          fileOrPattern,
          allMatches.size(),
          totalSize);

      filesSizeBytes.compareAndSet(null, totalSize);
      return totalSize;
    } else {
      long start = getStartOffset();
      long end = Math.min(getEndOffset(), getMaxEndOffset(options));
      return end - start;
    }
  }
@Test
  public void testEstimatedSizeOfFile() throws Exception {
    List<String> data = createStringDataset(3, 50);
    String fileName = "file";
    File file = createFileWithData(fileName, data);

    TestFileBasedSource source = new TestFileBasedSource(file.getPath(), 64, null);

    ExpectedLogs.LogSaver logSaver = new ExpectedLogs.LogSaver();
    LogManager.getLogManager().getLogger("").addHandler(logSaver);
    assertEquals(file.length(), source.getEstimatedSizeBytes(null));
    ExpectedLogs.verifyLogged(
        ExpectedLogs.matcher(
            Level.INFO, String.format("matched 1 files with total size %d", file.length())),
        logSaver);
    LogManager.getLogManager().getLogger("").removeHandler(logSaver);

    logSaver = new ExpectedLogs.LogSaver();
    LogManager.getLogManager().getLogger("").addHandler(logSaver);
    assertEquals(file.length(), source.getEstimatedSizeBytes(null));
    // Second call get result from cache and does not send match request
    ExpectedLogs.verifyNotLogged(
        ExpectedLogs.matcher(
            Level.INFO, String.format("matched 1 files with total size %d", file.length())),
        logSaver);
    LogManager.getLogManager().getLogger("").removeHandler(logSaver);
  }