dir-spooler-protolib/src/main/java/com/streamsets/pipeline/lib/dirspooler/SpoolDirUtil.java
public static DataParser getParser(
      WrappedFileSystem fs,
      WrappedFile file,
      DataFormat dataFormat,
      DataParserFactory parserFactory,
      String offset,
      int wholeFileMaxObjectLen,
      ELEval rateLimitElEval,
      ELVars rateLimitElVars,
      String rateLimit
  ) throws DataParserException, ELEvalException, IOException {
    DataParser parser;

    switch (dataFormat) {
      case WHOLE_FILE:
        FileRef fileRef = fs.getFileRefBuilder()
            .filePath(file.getAbsolutePath())
            .bufferSize(wholeFileMaxObjectLen)
            .rateLimit(FileRefUtil.evaluateAndGetRateLimit(rateLimitElEval, rateLimitElVars, rateLimit))
            .createMetrics(true)
            .totalSizeInBytes(file.getSize())
            .build();
        parser = parserFactory.getParser(file.getFileName(), file.getFileMetadata(), fileRef);
        break;
      default:
        parser = parserFactory.getParser(file.getFileName(), file.getInputStream(), offset);
    }

    return parser;
  }
public static DataParser getParser(
      WrappedFileSystem fs,
      WrappedFile file,
      DataFormat dataFormat,
      DataParserFactory parserFactory,
      String offset,
      int wholeFileMaxObjectLen,
      ELEval rateLimitElEval,
      ELVars rateLimitElVars,
      String rateLimit
  ) throws DataParserException, ELEvalException, IOException {
    DataParser parser;

    switch (dataFormat) {
      case WHOLE_FILE:
        FileRef fileRef = fs.getFileRefBuilder()
            .filePath(file.getAbsolutePath())
            .bufferSize(wholeFileMaxObjectLen)
            .rateLimit(FileRefUtil.evaluateAndGetRateLimit(rateLimitElEval, rateLimitElVars, rateLimit))
            .createMetrics(true)
            .totalSizeInBytes(file.getSize())
            .build();
        parser = parserFactory.getParser(file.getFileName(), file.getFileMetadata(), fileRef);
        break;
      default:
        parser = parserFactory.getParser(file.getFileName(), file.getInputStream(), offset);
    }

    return parser;
  }
/*
Returns the {@code DataParser} of the file, could be local file and hdfs file

@return  the {@code DataParser} of the file