public static SimpleFunction<Row, byte[]> getRowToJsonBytesFunction(Schema beamSchema) {
    return new RowToJsonFn<byte[]>(beamSchema) {
      @Override
      public byte[] apply(Row input) {
        String jsonString = RowJsonUtils.rowToJson(objectMapper, input);
        return jsonStringToByteArray(jsonString);
      }
    };
  }
@Test
  public void testGetRowToJsonBytesFunction() {
    for (TestCase<? extends RowEncodable> caze : testCases) {
      byte[] expected = caze.jsonBytes;
      byte[] actual = JsonUtils.getRowToJsonBytesFunction(caze.row.getSchema()).apply(caze.row);
      assertJsonEquals(caze.userT.toString(), expected, actual);
    }
  }