@Override
    public Object convert(String value) {
        if (value == null || value.isEmpty()) {
            return value;
        }
        final CSVParser parser = getCsvParser();
        final Map<String, String> fields = Maps.newHashMap();
        try {
            final String[] strings = parser.parseLine(value);
            if (strings.length != fieldNames.length) {
                LOG.error("Different number of columns in CSV data ({}) and configured field names ({}). Discarding input.",
                          strings.length, fieldNames.length);
                return null;
            }
            for (int i = 0; i < strings.length; i++) {
                fields.put(fieldNames[i], strings[i]);
            }
        } catch (IOException e) {
            LOG.error("Invalid CSV input, discarding input", e);
            return null;
        }
        return fields;
    }
@Test
    public void testSuccessfulConversion() throws ConfigurationException {
        Map<String, Object> configMap = Maps.newHashMap();
        configMap.put("column_header", "f1,f2");
        CsvConverter csvConverter = new CsvConverter(configMap);
        @SuppressWarnings("unchecked")
        Map<String, String> result = (Map<String, String>) csvConverter.convert("\"content1\",\"cont\\\\ent3\"");
        assertEquals("content1", result.get("f1"));
        assertEquals("cont\\ent3", result.get("f2"));
    }