### Question:
JsonRowSerializationSchema implements SerializationSchema<Row> { @Override public byte[] serialize(Row row) { if (node == null) { node = mapper.createObjectNode(); } try { convertRow(node, (RowTypeInfo) typeInfo, row); return mapper.writeValueAsBytes(node); } catch (Throwable t) { throw new RuntimeException("Could not serialize row '" + row + "'. " + "Make sure that the schema matches the input.", t); } } JsonRowSerializationSchema(TypeInformation<Row> typeInfo);  JsonRowSerializationSchema(String jsonSchema); @Override byte[] serialize(Row row);  }### Answer:
@Test public void testSerializationOfTwoRows() throws IOException { final TypeInformation<Row> rowSchema = Types.ROW_NAMED( new String[] {"f1", "f2", "f3"}, Types.INT, Types.BOOLEAN, Types.STRING); final Row row1 = new Row(3); row1.setField(0, 1); row1.setField(1, true); row1.setField(2, "str"); final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema); final JsonRowDeserializationSchema deserializationSchema = new JsonRowDeserializationSchema(rowSchema); byte[] bytes = serializationSchema.serialize(row1); assertEquals(row1, deserializationSchema.deserialize(bytes)); final Row row2 = new Row(3); row2.setField(0, 10); row2.setField(1, false); row2.setField(2, "newStr"); bytes = serializationSchema.serialize(row2); assertEquals(row2, deserializationSchema.deserialize(bytes)); }
@Test(expected = RuntimeException.class) public void testSerializeRowWithInvalidNumberOfFields() { final TypeInformation<Row> rowSchema = Types.ROW_NAMED( new String[] {"f1", "f2", "f3"}, Types.INT, Types.BOOLEAN, Types.STRING); final Row row = new Row(1); row.setField(0, 1); final JsonRowSerializationSchema serializationSchema = new JsonRowSerializationSchema(rowSchema); serializationSchema.serialize(row); }