@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
    @SuppressWarnings("unchecked")
    public void testEdgeCases() throws ConfigurationException {
        Map<String, Object> configMap = Maps.newHashMap();
        configMap.put("column_header", "f1,f2");
        CsvConverter csvConverter = new CsvConverter(configMap);
        String resultString = (String) csvConverter.convert("");
        assertEquals("", resultString);

        // too few fields
        Map<String, String> result = (Map<String, String>) csvConverter.convert("field1");
        assertNull("Too few fields in data doesn't work", result);

        // too many fields
        result = (Map<String, String>) csvConverter.convert("field1,field2,field3");
        assertNull("Too many fields in data doesn't work", result);

        // unclosed quote level
        result = (Map<String, String>) csvConverter.convert("field1,field2,\"field3");
        assertNull("Unbalanced quoting does not work", result);
    }