@Override
  public DynamicTableSink createDynamicTableSink(Context context) {
    Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions());
    checkArgument(!StringUtils.isNullOrEmpty(conf.getString(FlinkOptions.PATH)),
        "Option [path] should not be empty.");
    setupTableOptions(conf.getString(FlinkOptions.PATH), conf);
    ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
    sanityCheck(conf, schema);
    setupConfOptions(conf, context.getObjectIdentifier(), context.getCatalogTable(), schema);
    setupSortOptions(conf, context.getConfiguration());
    return new HoodieTableSink(conf, schema);
  }
@Test
  void testInferAvroSchemaForSink() {
    // infer the schema if not specified
    final HoodieTableSink tableSink1 =
        (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(this.conf));
    final Configuration conf1 = tableSink1.getConf();
    assertThat(conf1.get(FlinkOptions.SOURCE_AVRO_SCHEMA), is(INFERRED_SCHEMA));

    // set up the explicit schema using the file path
    this.conf.setString(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH, AVRO_SCHEMA_FILE_PATH);
    HoodieTableSink tableSink2 =
        (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(this.conf));
    Configuration conf2 = tableSink2.getConf();
    assertNull(conf2.get(FlinkOptions.SOURCE_AVRO_SCHEMA), "expect schema string as null");

    // infer special avro data types that needs namespace
    this.conf.removeConfig(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH);
    ResolvedSchema schema3 = SchemaBuilder.instance()
        .field("f_decimal", DataTypes.DECIMAL(3, 2).notNull())
        .field("f_map", DataTypes.MAP(DataTypes.VARCHAR(20), DataTypes.VARCHAR(10)))
        .field("f_array", DataTypes.ARRAY(DataTypes.VARCHAR(10)))
        .field("f_record", DataTypes.ROW(DataTypes.FIELD("r1", DataTypes.VARCHAR(10)), DataTypes.FIELD("r2", DataTypes.INT())))
        .primaryKey("f_decimal")
        .build();
    final HoodieTableSink tableSink3 =
        (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(MockContext.getInstance(this.conf, schema3, ""));
    final Configuration conf3 = tableSink3.getConf();
    final String expected = AvroSchemaConverter.convertToSchema(schema3.toSinkRowDataType().getLogicalType(), AvroSchemaUtils.getAvroRecordQualifiedName("t1")).toString();
    assertThat(conf3.get(FlinkOptions.SOURCE_AVRO_SCHEMA), is(expected));
  }