### Question:
KsqlDelimitedDeserializer implements Deserializer<GenericRow> { @Override public GenericRow deserialize(final String topic, final byte[] bytes) { if (bytes == null) { return null; } String recordCsvString = new String(bytes, StandardCharsets.UTF_8); try { List<CSVRecord> csvRecords = CSVParser.parse(recordCsvString, CSVFormat.DEFAULT).getRecords(); if (csvRecords == null || csvRecords.isEmpty()) { throw new KsqlException("Deserialization error in the delimited line: " + recordCsvString); } CSVRecord csvRecord = csvRecords.get(0); if (csvRecord == null || csvRecord.size() == 0) { throw new KsqlException("Deserialization error in the delimited line: " + recordCsvString); } List<Object> columns = new ArrayList<>(); if (csvRecord.size() != schema.fields().size()) { throw new KsqlException( String.format( "Unexpected field count, csvFields:%d schemaFields:%d line: %s", csvRecord.size(), schema.fields().size(), recordCsvString ) ); } for (int i = 0; i < csvRecord.size(); i++) { if (csvRecord.get(i) == null) { columns.add(null); } else { columns.add(enforceFieldType(schema.fields().get(i).schema(), csvRecord.get(i))); } } return new GenericRow(columns); } catch (Exception e) { throw new SerializationException( "Exception in deserializing the delimited row: " + recordCsvString, e ); } } KsqlDelimitedDeserializer(Schema schema); @Override void configure(Map<String, ?> map, boolean b); @Override GenericRow deserialize(final String topic, final byte[] bytes); @Override void close();  }### Answer:
@Test public void shouldDeserializeDelimitedCorrectly() { String rowString = "1511897796092,1,item_1,10.0\r\n"; KsqlDelimitedDeserializer ksqlJsonDeserializer = new KsqlDelimitedDeserializer(orderSchema); GenericRow genericRow = ksqlJsonDeserializer.deserialize("", rowString.getBytes()); assertThat(genericRow.getColumns().size(), equalTo(4)); assertThat((Long) genericRow.getColumns().get(0), equalTo(1511897796092L)); assertThat((Long) genericRow.getColumns().get(1), equalTo(1L)); assertThat((String) genericRow.getColumns().get(2), equalTo("item_1")); assertThat((Double) genericRow.getColumns().get(3), equalTo(10.0)); }
@Test public void shouldDeserializeJsonCorrectlyWithRedundantFields() throws JsonProcessingException { String rowString = "1511897796092,1,item_1,\r\n"; KsqlDelimitedDeserializer ksqlJsonDeserializer = new KsqlDelimitedDeserializer(orderSchema); GenericRow genericRow = ksqlJsonDeserializer.deserialize("", rowString.getBytes()); assertThat(genericRow.getColumns().size(), equalTo(4)); assertThat((Long) genericRow.getColumns().get(0), equalTo(1511897796092L)); assertThat((Long) genericRow.getColumns().get(1), equalTo(1L)); assertThat((String) genericRow.getColumns().get(2), equalTo("item_1")); Assert.assertNull(genericRow.getColumns().get(3)); }