### Question:
KsqlJsonSerializer implements Serializer<GenericRow> { @Override public byte[] serialize(final String topic, final GenericRow data) { if (data == null) { return null; } try { return objectMapper.writeValueAsBytes(dataToMap(data)); } catch (Exception e) { throw new SerializationException("Error serializing JSON message", e); } } KsqlJsonSerializer(Schema schema); @SuppressWarnings("unchecked") @Override void configure(final Map<String, ?> props, final boolean isKey); @Override byte[] serialize(final String topic, final GenericRow data); @Override void close();  }### Answer:
@Test public void shouldSerializeRowCorrectly() { List columns = Arrays.asList(1511897796092L, 1L, "item_1", 10.0, new Double[]{100.0}, Collections.singletonMap("key1", 100.0)); GenericRow genericRow = new GenericRow(columns); KsqlJsonSerializer ksqlJsonDeserializer = new KsqlJsonSerializer(orderSchema); byte[] bytes = ksqlJsonDeserializer.serialize("t1", genericRow); String jsonString = new String(bytes); assertThat("Incorrect serialization.", jsonString, equalTo("{\"ORDERID\":1,\"ARRAYCOL\":[100.0],\"ORDERUNITS\":10.0,\"ORDERTIME\":1511897796092,\"MAPCOL\":{\"key1\":100.0},\"ITEMID\":\"item_1\"}")); }
@Test public void shouldSerializeRowWithNull() { List columns = Arrays.asList(1511897796092L, 1L, "item_1", 10.0, null, null); GenericRow genericRow = new GenericRow(columns); KsqlJsonSerializer ksqlJsonDeserializer = new KsqlJsonSerializer(orderSchema); byte[] bytes = ksqlJsonDeserializer.serialize("t1", genericRow); String jsonString = new String(bytes); assertThat("Incorrect serialization.", jsonString, equalTo("{\"ORDERID\":1,\"ARRAYCOL\":null,\"ORDERUNITS\":10.0,\"ORDERTIME\":1511897796092,\"MAPCOL\":null,\"ITEMID\":\"item_1\"}")); }