@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 shouldThrowOnInvalidDate() {
    // Given:
    givenSingleColumnSerializer(SqlTypes.DATE);

    final List<?> values = Collections.singletonList(new Date(1234));

    // Then:
    final Exception e = assertThrows(
        SerializationException.class,
        () -> serializer.serialize("", values)
    );
    assertThat(e.getCause().getMessage(),
        is("Date type should not have any time fields set to non-zero values."));
  }