public BatchConfig(String tableNameWithType, Map<String, String> batchConfigsMap) {
    _batchConfigMap = batchConfigsMap;
    _tableNameWithType = tableNameWithType;

    String inputFormat = batchConfigsMap.get(BatchConfigProperties.INPUT_FORMAT);
    if (inputFormat != null) {
      _inputFormat = FileFormat.valueOf(inputFormat.toUpperCase());
    } else {
      _inputFormat = null;
    }
    _inputDirURI = batchConfigsMap.get(BatchConfigProperties.INPUT_DIR_URI);
    _inputFsClassName = batchConfigsMap.get(BatchConfigProperties.INPUT_FS_CLASS);
    _inputFsProps =
        IngestionConfigUtils.extractPropsMatchingPrefix(batchConfigsMap, BatchConfigProperties.INPUT_FS_PROP_PREFIX);

    _outputDirURI = batchConfigsMap.get(BatchConfigProperties.OUTPUT_DIR_URI);
    _outputFsClassName = batchConfigsMap.get(BatchConfigProperties.OUTPUT_FS_CLASS);
    _outputFsProps =
        IngestionConfigUtils.extractPropsMatchingPrefix(batchConfigsMap, BatchConfigProperties.OUTPUT_FS_PROP_PREFIX);
    _overwriteOutput = Boolean.parseBoolean(batchConfigsMap.get(BatchConfigProperties.OVERWRITE_OUTPUT));

    _recordReaderClassName = batchConfigsMap.get(BatchConfigProperties.RECORD_READER_CLASS);
    _recordReaderConfigClassName = batchConfigsMap.get(BatchConfigProperties.RECORD_READER_CONFIG_CLASS);
    _recordReaderProps = IngestionConfigUtils.extractPropsMatchingPrefix(batchConfigsMap,
        BatchConfigProperties.RECORD_READER_PROP_PREFIX);

    _segmentNameGeneratorType = IngestionConfigUtils.getSegmentNameGeneratorType(batchConfigsMap);
    _segmentNameGeneratorConfigs = IngestionConfigUtils.extractPropsMatchingPrefix(batchConfigsMap,
        BatchConfigProperties.SEGMENT_NAME_GENERATOR_PROP_PREFIX);
    Map<String, String> segmentNameGeneratorProps = IngestionConfigUtils.getSegmentNameGeneratorProps(batchConfigsMap);
    _segmentName = segmentNameGeneratorProps.get(BatchConfigProperties.SEGMENT_NAME);
    _segmentNamePrefix = segmentNameGeneratorProps.get(BatchConfigProperties.SEGMENT_NAME_PREFIX);
    _segmentNamePostfix = segmentNameGeneratorProps.get(BatchConfigProperties.SEGMENT_NAME_POSTFIX);
    _excludeSequenceId = Boolean.parseBoolean(segmentNameGeneratorProps.get(BatchConfigProperties.EXCLUDE_SEQUENCE_ID));
    _sequenceId = batchConfigsMap.get(BatchConfigProperties.SEQUENCE_ID);
    _appendUUIDToSegmentName =
        Boolean.parseBoolean(segmentNameGeneratorProps.get(BatchConfigProperties.APPEND_UUID_TO_SEGMENT_NAME));
    _excludeTimeInSegmentName =
        Boolean.parseBoolean(segmentNameGeneratorProps.get(BatchConfigProperties.EXCLUDE_TIME_IN_SEGMENT_NAME));

    _pushMode = IngestionConfigUtils.getPushMode(batchConfigsMap);
    _pushAttempts = IngestionConfigUtils.getPushAttempts(batchConfigsMap);
    _pushParallelism = IngestionConfigUtils.getPushParallelism(batchConfigsMap);
    _pushIntervalRetryMillis = IngestionConfigUtils.getPushRetryIntervalMillis(batchConfigsMap);
    _pushSegmentURIPrefix = batchConfigsMap.get(BatchConfigProperties.PUSH_SEGMENT_URI_PREFIX);
    _pushSegmentURISuffix = batchConfigsMap.get(BatchConfigProperties.PUSH_SEGMENT_URI_SUFFIX);
    _pushControllerURI = batchConfigsMap.get(BatchConfigProperties.PUSH_CONTROLLER_URI);
    _outputSegmentDirURI = batchConfigsMap.get(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI);
  }
@Test
  public void testBatchConfig() {
    Map<String, String> batchConfigMap = new HashMap<>();
    String tableName = "foo_REALTIME";
    String inputDir = "s3://foo/input";
    String outputDir = "s3://foo/output";
    String inputFsClass = "org.apache.S3FS";
    String outputFsClass = "org.apache.GcsGS";
    String region = "us-west";
    String username = "foo";
    String accessKey = "${ACCESS_KEY}";
    String secretKey = "${SECRET_KEY}";
    String inputFormat = "csv";
    String recordReaderClass = "org.foo.CSVRecordReader";
    String recordReaderConfigClass = "org.foo.CSVRecordReaderConfig";
    String separator = "|";
    batchConfigMap.put(BatchConfigProperties.INPUT_DIR_URI, inputDir);
    batchConfigMap.put(BatchConfigProperties.OUTPUT_DIR_URI, outputDir);
    batchConfigMap.put(BatchConfigProperties.INPUT_FS_CLASS, inputFsClass);
    batchConfigMap.put(BatchConfigProperties.OUTPUT_FS_CLASS, outputFsClass);
    batchConfigMap.put(BatchConfigProperties.INPUT_FORMAT, inputFormat);
    batchConfigMap.put(BatchConfigProperties.RECORD_READER_CLASS, recordReaderClass);
    batchConfigMap.put(BatchConfigProperties.RECORD_READER_CONFIG_CLASS, recordReaderConfigClass);
    batchConfigMap.put(BatchConfigProperties.INPUT_FS_PROP_PREFIX + ".region", region);
    batchConfigMap.put(BatchConfigProperties.INPUT_FS_PROP_PREFIX + ".username", username);
    batchConfigMap.put(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".region", region);
    batchConfigMap.put(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".accessKey", accessKey);
    batchConfigMap.put(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".secretKey", secretKey);
    batchConfigMap.put(BatchConfigProperties.RECORD_READER_PROP_PREFIX + ".separator", separator);

    // config with all the right properties
    BatchConfig batchConfig = new BatchConfig(tableName, batchConfigMap);
    assertEquals(batchConfig.getInputDirURI(), inputDir);
    assertEquals(batchConfig.getOutputDirURI(), outputDir);
    assertEquals(batchConfig.getInputFsClassName(), inputFsClass);
    assertEquals(batchConfig.getOutputFsClassName(), outputFsClass);
    assertEquals(batchConfig.getInputFormat(), FileFormat.CSV);
    assertEquals(batchConfig.getRecordReaderClassName(), recordReaderClass);
    assertEquals(batchConfig.getRecordReaderConfigClassName(), recordReaderConfigClass);
    assertEquals(batchConfig.getInputFsProps().size(), 2);
    assertEquals(batchConfig.getInputFsProps().get(BatchConfigProperties.INPUT_FS_PROP_PREFIX + ".region"), region);
    assertEquals(batchConfig.getInputFsProps().get(BatchConfigProperties.INPUT_FS_PROP_PREFIX + ".username"), username);
    assertEquals(batchConfig.getOutputFsProps().size(), 3);
    assertEquals(batchConfig.getOutputFsProps().get(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".region"), region);
    assertEquals(batchConfig.getOutputFsProps().get(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".accessKey"),
        accessKey);
    assertEquals(batchConfig.getOutputFsProps().get(BatchConfigProperties.OUTPUT_FS_PROP_PREFIX + ".secretKey"),
        secretKey);
    assertEquals(batchConfig.getRecordReaderProps().size(), 1);
    assertEquals(batchConfig.getRecordReaderProps().get(BatchConfigProperties.RECORD_READER_PROP_PREFIX + ".separator"),
        separator);
    assertEquals(batchConfig.getTableNameWithType(), tableName);
  }