public static SimpleFunction<byte[], Row> getJsonBytesToRowFunction(Schema beamSchema) {
    return new JsonToRowFn<byte[]>(beamSchema) {
      @Override
      public Row apply(byte[] input) {
        String jsonString = byteArrayToJsonString(input);
        return RowJsonUtils.jsonToRow(objectMapper, jsonString);
      }
    };
  }
@Test
  public void testGetJsonBytesToRowFunction() {
    for (TestCase<? extends RowEncodable> caze : testCases) {
      Row expected = caze.row;
      Row actual = JsonUtils.getJsonBytesToRowFunction(expected.getSchema()).apply(caze.jsonBytes);
      assertEquals(caze.userT.toString(), expected, actual);
    }
  }