class CSVRecordReader extends AbstractCSVRecordReader {

    @Override
    public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException {
        final RecordSchema schema = getSchema();

        final List<RecordField> recordFields = getRecordFields();
        final int numFieldNames = recordFields.size();

        for (final CSVRecord csvRecord : csvParser) {
            final Map<String, Object> values = new LinkedHashMap<>(recordFields.size() * 2);
            for (int i = 0; i < csvRecord.size(); i++) {
                final String rawValue = csvRecord.get(i);

                final String rawFieldName;
                final DataType dataType;
                if (i >= numFieldNames) {
                    if (!dropUnknownFields) {
                        values.put("unknown_field_index_" + i, rawValue);
                    }

                    continue;
                } else {
                    final RecordField recordField = recordFields.get(i);
                    rawFieldName = recordField.getFieldName();
                    dataType = recordField.getDataType();
                }


                final Object value;
                if (coerceTypes) {
                    value = convert(rawValue, dataType, rawFieldName);
                } else {
                    // The CSV Reader is going to return all fields as Strings, because CSV doesn't have any way to
                    // dictate a field type. As a result, we will use the schema that we have to attempt to convert
                    // the value into the desired type if it's a simple type.
                    value = convertSimpleIfPossible(rawValue, dataType, rawFieldName);
                }

                values.put(rawFieldName, value);
            }

            return new MapRecord(schema, values, coerceTypes, dropUnknownFields);
        }

        return null;
    }

    public  CSVRecordReader(final InputStream in, final ComponentLog logger, final RecordSchema schema, final CSVFormat csvFormat, final boolean hasHeader, final boolean ignoreHeader,
                           final String dateFormat, final String timeFormat, final String timestampFormat, final String encoding);

}