@Override
  public PayloadSerializer getSerializer(Schema schema, Map<String, Object> tableParams) {
    Class<? extends TBase> thriftClass = getMessageClass(tableParams);
    TProtocolFactory protocolFactory = getProtocolFactory(tableParams);
    inferAndVerifySchema(thriftClass, schema);
    return getPayloadSerializer(schema, protocolFactory, thriftClass);
  }
@Test
  public void invalidArgs() {
    assertThrows(
        IllegalArgumentException.class,
        () -> provider.getSerializer(SHUFFLED_SCHEMA, ImmutableMap.of()));
    assertThrows(
        IllegalArgumentException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of("thriftClass", "", "thriftProtocolFactoryClass", "")));
    assertThrows(
        IllegalArgumentException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of(
                    "thriftClass",
                    "",
                    "thriftProtocolFactoryClass",
                    TCompactProtocol.Factory.class.getName())));
    assertThrows(
        IllegalArgumentException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of(
                    "thriftClass",
                    TestThriftMessage.class.getName(),
                    "thriftProtocolFactoryClass",
                    "")));
    assertThrows(
        ClassCastException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of(
                    "thriftClass", ImmutableList.class.getName(),
                    "thriftProtocolFactoryClass", TCompactProtocol.Factory.class.getName())));
    assertThrows(
        ClassCastException.class,
        () ->
            provider.getSerializer(
                SHUFFLED_SCHEMA,
                ImmutableMap.of(
                    "thriftClass", TestThriftMessage.class.getName(),
                    "thriftProtocolFactoryClass", ImmutableList.class.getName())));
    assertThrows(
        IllegalArgumentException.class,
        () ->
            provider.getSerializer(
                Schema.builder()
                    .addStringField("f_NOTACTUALLYINMESSAGE")
                    .addInt32Field("f_int")
                    .addArrayField("f_double_array", Schema.FieldType.DOUBLE)
                    .addDoubleField("f_double")
                    .addInt64Field("f_long")
                    .build(),
                ImmutableMap.of(
                    "thriftClass", TestThriftMessage.class.getName(),
                    "thriftProtocolFactoryClass", TCompactProtocol.Factory.class.getName())));
  }