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

    try {
      final StringWriter stringWriter = new StringWriter();
      final CSVPrinter csvPrinter = new CSVPrinter(stringWriter, csvFormat);
      csvPrinter.printRecord(() -> new FieldIterator(data, schema));
      final String result = stringWriter.toString();
      return result.substring(0, result.length() - 2).getBytes(StandardCharsets.UTF_8);
    } catch (final Exception e) {
      throw new SerializationException("Error serializing CSV message", e);
    }
  }
@Test
  public void shouldSerializeRowCorrectly() {
    // Given:
    final List<?> values = Arrays.asList(1511897796092L, 1L, "item_1", 10.0, new Time(100), new Date(864000000L), new Timestamp(100),
        ByteBuffer.wrap(new byte[] {123}), new BigDecimal("10"));

    // When:
    final byte[] bytes = serializer.serialize("t1", values);

    // Then:
    final String delimitedString = new String(bytes, StandardCharsets.UTF_8);
    assertThat(delimitedString, equalTo("1511897796092,1,item_1,10.0,100,10,100,ew==,10"));
  }