@Override
  public List<?> deserialize(final String topic, final byte[] bytes) {
    if (bytes == null) {
      return null;
    }

    try {
      final String recordCsvString = new String(bytes, StandardCharsets.UTF_8);
      final List<CSVRecord> csvRecords = CSVParser.parse(recordCsvString, csvFormat)
          .getRecords();

      if (csvRecords.isEmpty()) {
        throw new SerializationException("No fields in record");
      }

      final CSVRecord csvRecord = csvRecords.get(0);
      if (csvRecord == null || csvRecord.size() == 0) {
        throw new SerializationException("No fields in record.");
      }

      SerdeUtils.throwOnColumnCountMismatch(parsers.size(), csvRecord.size(), false, topic);

      final List<Object> values = new ArrayList<>(parsers.size());
      final Iterator<Parser> pIt = parsers.iterator();

      for (int i = 0; i < csvRecord.size(); i++) {
        final String value = csvRecord.get(i);
        final Parser parser = pIt.next();

        final Object parsed = value == null || value.isEmpty()
            ? null
            : parser.parse(value);

        values.add(parsed);
      }

      return values;
    } catch (final Exception e) {
      throw new SerializationException("Error deserializing delimited", e);
    }
  }
@Test
  public void shouldThrowIfRowHasTooMayColumns() {
    // Given:
    final byte[] bytes = "1511897796092,1,item_1,10.0,10.10,100,100,100,100,extra\r\n"
        .getBytes(StandardCharsets.UTF_8);

    // When:
    final Exception e = assertThrows(
        SerializationException.class,
        () -> deserializer.deserialize("t", bytes)
    );

    // Then:
    assertThat(e.getCause().getMessage(),
        is("Column count mismatch on deserialization. topic: t, expected: 9, got: 10"));
  }