### Question:
KsqlDelimitedSerializer implements Serializer<GenericRow> { @Override public byte[] serialize(final String topic, final GenericRow genericRow) { if (genericRow == null) { return null; } try { StringWriter stringWriter = new StringWriter(); CSVPrinter csvPrinter = new CSVPrinter(stringWriter, CSVFormat.DEFAULT); csvPrinter.printRecord(genericRow.getColumns()); String result = stringWriter.toString(); return result.substring(0, result.length() - 2).getBytes(StandardCharsets.UTF_8); } catch (Exception e) { throw new SerializationException("Error serializing CSV message", e); } } KsqlDelimitedSerializer(Schema schema); @Override void configure(Map<String, ?> map, boolean b); @Override byte[] serialize(final String topic, final GenericRow genericRow); @Override void close();  }### Answer:
@Test public void shouldSerializeRowCorrectly() { List columns = Arrays.asList(1511897796092L, 1L, "item_1", 10.0); GenericRow genericRow = new GenericRow(columns); KsqlDelimitedSerializer ksqlDelimitedSerializer = new KsqlDelimitedSerializer(orderSchema); byte[] bytes = ksqlDelimitedSerializer.serialize("t1", genericRow); String delimitedString = new String(bytes); assertThat("Incorrect serialization.", delimitedString, equalTo("1511897796092,1,item_1,10.0")); }
@Test public void shouldSerializeRowWithNull() { List columns = Arrays.asList(1511897796092L, 1L, "item_1", null); GenericRow genericRow = new GenericRow(columns); KsqlDelimitedSerializer ksqlDelimitedSerializer = new KsqlDelimitedSerializer(orderSchema); byte[] bytes = ksqlDelimitedSerializer.serialize("t1", genericRow); String delimitedString = new String(bytes); assertThat("Incorrect serialization.", delimitedString, equalTo("1511897796092,1,item_1,")); }