code: protected void init(SourceState state) {
      retriever.init(state);
  
      try {
        initFileSystemHelper(state);
      } catch (FileBasedHelperException e) {
        Throwables.propagate(e);
      }
  
      AvroFsHelper fsHelper = (AvroFsHelper) this.fsHelper;
      this.fs = fsHelper.getFileSystem();
  
      this.sourceState = state;
  
      this.lowWaterMark =
          getLowWaterMark(state.getPreviousWorkUnitStates(), state.getProp(DATE_PARTITIONED_SOURCE_MIN_WATERMARK_VALUE,
              String.valueOf(DEFAULT_DATE_PARTITIONED_SOURCE_MIN_WATERMARK_VALUE)));
  
      this.maxFilesPerJob = state.getPropAsInt(DATE_PARTITIONED_SOURCE_MAX_FILES_PER_JOB,
          DEFAULT_DATE_PARTITIONED_SOURCE_MAX_FILES_PER_JOB);
  
      this.maxWorkUnitsPerJob = state.getPropAsInt(DATE_PARTITIONED_SOURCE_MAX_WORKUNITS_PER_JOB,
          DEFAULT_DATE_PARTITIONED_SOURCE_MAX_WORKUNITS_PER_JOB);
  
      this.tableType = TableType.valueOf(state.getProp(ConfigurationKeys.EXTRACT_TABLE_TYPE_KEY).toUpperCase());
  
      this.fileCount = 0;
  
      this.sourceDir = new Path(state.getProp(ConfigurationKeys.SOURCE_FILEBASED_DATA_DIRECTORY));
   }
docs: Gobblin calls the {@link Source#getWorkunits(SourceState)} method after creating a {@link Source} object with a
  blank constructor, so any custom initialization of the object needs to be done here.