@Override
  public PayloadSerializer getSerializer(Schema schema, Map<String, Object> tableParams) {
    Class<? extends Message> protoClass = getClass(tableParams);
    inferAndVerifySchema(protoClass, schema);
    SimpleFunction<byte[], Row> toRowFn = ProtoMessageSchema.getProtoBytesToRowFn(protoClass);
    return PayloadSerializer.of(
        ProtoMessageSchema.getRowToProtoBytesFn(protoClass),
        bytes -> {
          Row rawRow = toRowFn.apply(bytes);
          return castRow(rawRow, rawRow.getSchema(), schema);
        });
  }
@Test
  public void serialize() throws Exception {
    byte[] bytes =
        provider
            .getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of("protoClass", PayloadMessages.TestMessage.class.getName()))
            .serialize(ROW);
    PayloadMessages.TestMessage result = PayloadMessages.TestMessage.parseFrom(bytes);
    assertEquals(MESSAGE, result);
  }