code: @Override
    public void sourceConfInit(FlowProcess<JobConf> process, JobConf conf) {
      try {
        Path root = getQualifiedPath(conf);
        if (_options.attrs != null && _options.attrs.length > 0) {
          Pail pail = new Pail(_pailRoot, conf);
          for (List<String> attr : _options.attrs) {
            String rel = Utils.join(attr, "/");
            pail.getSubPail(rel); //ensure the path exists
            Path toAdd = new Path(root, rel);
            LOG.info("Adding input path " + toAdd.toString());
            FileInputFormat.addInputPath(conf, toAdd);
          }
        } else {
          FileInputFormat.addInputPath(conf, root);
        }
  
        getScheme().sourceConfInit(process, this, conf);
        makeLocal(conf, getQualifiedPath(conf), "forcing job to local mode, via source: ");
        TupleSerialization.setSerializations(conf);
      } catch (IOException e) {
        throw new TapException(e);
      }
    }
docs: no good way to override this, just had to copy/paste and modify