@Override
    public void sourceConfInit(FlowProcess<JobConf> flowProcess,
            Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {

        conf.setInputFormat(AccumuloInputFormat.class);
        setFields(TapType.SOURCE);

        AccumuloTap accumuloTap = (AccumuloTap) tap;

        if (false == ConfiguratorBase.isConnectorInfoSet(
                AccumuloInputFormat.class, conf)) {
            try {
                AccumuloInputFormat.setConnectorInfo(conf,
                        accumuloTap.accumuloUserName,
                        accumuloTap.accumuloAuthToken);
                AccumuloInputFormat.setScanAuthorizations(conf,
                        accumuloTap.accumuloAuthorizations);
                AccumuloInputFormat.setInputTableName(conf,
                        accumuloTap.tableName);
                AccumuloInputFormat.setZooKeeperInstance(conf,
                        accumuloTap.accumuloInstanceName,
                        accumuloTap.accumuloZookeeperQuorum);
            } catch (AccumuloSecurityException ex) {
                LOG.error("Error in AccumuloScheme.sourceConfInit", ex);
            }
            if (columnFamilyColumnQualifierPairs.size() > 0) {
                AccumuloInputFormat.fetchColumns(conf,
                        columnFamilyColumnQualifierPairs);
            }

            if (!rowKeyStart.equals("*") && !rowKeyEnd.equals("*/0")) {
                AccumuloInputFormat.setRanges(conf,
                        Collections.singleton(new Range(rowKeyStart, rowKeyEnd)));
            }

            if (addRegexFilter) {
                IteratorSetting regex = new IteratorSetting(50, "regex",
                        RegExFilter.class);
                RegExFilter.setRegexs(regex, this.rowRegex,
                        this.columnFamilyRegex, this.columnQualifierRegex,
                        this.valueRegex, false);
                AccumuloInputFormat.addIterator(conf, regex);
            }

            if (maxAge != null) {
              IteratorSetting maxAgeFilter = new IteratorSetting(48, "maxAgeFilter",
                  AgeOffFilter.class);
              AgeOffFilter.setTTL(maxAgeFilter, maxAge);
              AccumuloInputFormat.addIterator(conf, maxAgeFilter);
            }

            if (minAge != null) {
              IteratorSetting minAgeFilter = new IteratorSetting(49, "minAgeFilter",
                  AgeOffFilter.class);
              AgeOffFilter.setTTL(minAgeFilter, minAge + 1);
              AgeOffFilter.setNegate(minAgeFilter, true);
              AccumuloInputFormat.addIterator(conf, minAgeFilter);
            }
        }
    }