@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 invalidArgs() {
    assertThrows(
        IllegalArgumentException.class,
        () -> provider.getSerializer(SHUFFLED_SCHEMA, ImmutableMap.of()));
    assertThrows(
        IllegalArgumentException.class,
        () -> provider.getSerializer(SHUFFLED_SCHEMA, ImmutableMap.of("protoClass", "")));
    assertThrows(
        ClassCastException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA, ImmutableMap.of("protoClass", ImmutableList.class.getName())));
    assertThrows(
        IllegalArgumentException.class,
        () ->
            provider.getSerializer(
                Schema.builder()
                    .addStringField("f_NOTACTUALLYINMESSAGE")
                    .addInt32Field("f_int")
                    .addArrayField("f_float_array", FieldType.FLOAT)
                    .addDoubleField("f_double")
                    .addInt64Field("f_long")
                    .build(),
                ImmutableMap.of("protoClass", PayloadMessages.TestMessage.class.getName())));
  }