@Override
  public void init(File dataFile, @Nullable Set<String> fieldsToRead, @Nullable RecordReaderConfig recordReaderConfig)
      throws IOException {
    _dataFile = dataFile;
    CSVRecordReaderConfig config = (CSVRecordReaderConfig) recordReaderConfig;
    Character multiValueDelimiter = null;
    if (config == null) {
      _format = CSVFormat.DEFAULT.builder().setDelimiter(CSVRecordReaderConfig.DEFAULT_DELIMITER).setHeader().build();
      multiValueDelimiter = CSVRecordReaderConfig.DEFAULT_MULTI_VALUE_DELIMITER;
    } else {
      CSVFormat format;
      String formatString = config.getFileFormat();
      if (formatString == null) {
        format = CSVFormat.DEFAULT;
      } else {
        switch (formatString.toUpperCase()) {
          case "EXCEL":
            format = CSVFormat.EXCEL;
            break;
          case "MYSQL":
            format = CSVFormat.MYSQL;
            break;
          case "RFC4180":
            format = CSVFormat.RFC4180;
            break;
          case "TDF":
            format = CSVFormat.TDF;
            break;
          default:
            format = CSVFormat.DEFAULT;
            break;
        }
      }
      char delimiter = config.getDelimiter();
      format = format.builder().setDelimiter(delimiter).build();

      if (config.isSkipUnParseableLines()) {
        _useLineIterator = true;
      }

      _isHeaderProvided = config.getHeader() != null;
      _skipHeaderRecord = config.isSkipHeader();
      _format = format.builder()
          .setHeader()
          .setSkipHeaderRecord(config.isSkipHeader())
          .setCommentMarker(config.getCommentMarker())
          .setEscape(config.getEscapeCharacter())
          .setIgnoreEmptyLines(config.isIgnoreEmptyLines())
          .setIgnoreSurroundingSpaces(config.isIgnoreSurroundingSpaces())
          .setQuote(config.getQuoteCharacter())
          .build();

      if (config.getQuoteMode() != null) {
        _format = _format.builder().setQuoteMode(QuoteMode.valueOf(config.getQuoteMode())).build();
      }

      if (config.getRecordSeparator() != null) {
        _format = _format.builder().setRecordSeparator(config.getRecordSeparator()).build();
      }

      String nullString = config.getNullStringValue();
      if (nullString != null) {
        _format = _format.builder().setNullString(nullString).build();
      }

      if (_isHeaderProvided) {
        _headerMap = parseLineAsHeader(config.getHeader());
        _format = _format.builder().setHeader(_headerMap.keySet().toArray(new String[0])).build();
        if (!_useLineIterator) {
          validateHeaderForDelimiter(delimiter, config.getHeader(), _format);
        }
      }

      if (config.isMultiValueDelimiterEnabled()) {
        multiValueDelimiter = config.getMultiValueDelimiter();
      }
    }
    _recordExtractor = new CSVRecordExtractor();

    init();

    CSVRecordExtractorConfig recordExtractorConfig = new CSVRecordExtractorConfig();
    recordExtractorConfig.setMultiValueDelimiter(multiValueDelimiter);
    recordExtractorConfig.setColumnNames(_headerMap.keySet());
    _recordExtractor.init(fieldsToRead, recordExtractorConfig);
  }
@Test
  public void testInvalidDelimiterInHeader() {
    // setup
    CSVRecordReaderConfig csvRecordReaderConfig = new CSVRecordReaderConfig();
    csvRecordReaderConfig.setMultiValueDelimiter(CSV_MULTI_VALUE_DELIMITER);
    csvRecordReaderConfig.setHeader("col1;col2;col3;col4;col5;col6;col7;col8;col9;col10");
    csvRecordReaderConfig.setDelimiter(',');
    CSVRecordReader csvRecordReader = new CSVRecordReader();

    //execute and assert
    Assert.assertThrows(IllegalArgumentException.class,
        () -> csvRecordReader.init(_dataFile, null, csvRecordReaderConfig));
  }