@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 shouldSerializeRowCorrectlyWithDifferentDelimiter() {
    // 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"));

    final KsqlDelimitedSerializer serializer1 =
        new KsqlDelimitedSerializer(SCHEMA, CSVFormat.DEFAULT.withDelimiter('\t'));

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

    // Then:
    assertThat(new String(bytes, StandardCharsets.UTF_8), is("1511897796092\t1\titem_1\t10.0\t100\t10\t100\tew==\t10"));
  }