@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 testSetupHoodieKeyOptionsForSink() {
    this.conf.setString(FlinkOptions.RECORD_KEY_FIELD, "dummyField");
    this.conf.setString(FlinkOptions.KEYGEN_CLASS_NAME, "dummyKeyGenClass");
    // definition with simple primary key and partition path
    ResolvedSchema schema1 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.BIGINT())
        .field("ts", DataTypes.TIMESTAMP(3))
        .primaryKey("f0")
        .build();
    final MockContext sinkContext1 = MockContext.getInstance(this.conf, schema1, "f2");
    final HoodieTableSink tableSink1 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sinkContext1);
    final Configuration conf1 = tableSink1.getConf();
    assertThat(conf1.get(FlinkOptions.RECORD_KEY_FIELD), is("f0"));
    assertThat(conf1.get(FlinkOptions.KEYGEN_CLASS_NAME), is("dummyKeyGenClass"));

    // definition with complex primary keys and partition paths
    this.conf.removeConfig(FlinkOptions.KEYGEN_CLASS_NAME);
    ResolvedSchema schema2 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20).notNull())
        .field("f2", DataTypes.TIMESTAMP(3))
        .field("ts", DataTypes.TIMESTAMP(3))
        .primaryKey("f0", "f1")
        .build();
    final MockContext sinkContext2 = MockContext.getInstance(this.conf, schema2, "f2");
    final HoodieTableSink tableSink2 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sinkContext2);
    final Configuration conf2 = tableSink2.getConf();
    assertThat(conf2.get(FlinkOptions.RECORD_KEY_FIELD), is("f0,f1"));
    assertThat(conf2.get(FlinkOptions.KEYGEN_CLASS_NAME), is(ComplexAvroKeyGenerator.class.getName()));

    // definition with complex primary keys and empty partition paths
    this.conf.removeConfig(FlinkOptions.KEYGEN_CLASS_NAME);
    final MockContext sinkContext3 = MockContext.getInstance(this.conf, schema2, "");
    final HoodieTableSink tableSink3 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sinkContext3);
    final Configuration conf3 = tableSink3.getConf();
    assertThat(conf3.get(FlinkOptions.RECORD_KEY_FIELD), is("f0,f1"));
    assertThat(conf3.get(FlinkOptions.KEYGEN_CLASS_NAME), is(NonpartitionedAvroKeyGenerator.class.getName()));

    // definition of bucket index
    this.conf.setString(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BUCKET.name());
    final MockContext sinkContext4 = MockContext.getInstance(this.conf, schema2, "");
    final HoodieTableSink tableSink4 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sinkContext4);
    final Configuration conf4 = tableSink4.getConf();
    assertThat(conf4.get(FlinkOptions.RECORD_KEY_FIELD), is("f0,f1"));
    assertThat(conf4.get(FlinkOptions.INDEX_KEY_FIELD), is("f0,f1"));
    assertThat(conf4.get(FlinkOptions.INDEX_TYPE), is(HoodieIndex.IndexType.BUCKET.name()));
    assertThat(conf4.get(FlinkOptions.KEYGEN_CLASS_NAME), is(NonpartitionedAvroKeyGenerator.class.getName()));
  }