@Override
  public Set<ConfigOption<?>> requiredOptions() {
    return Collections.singleton(FlinkOptions.PATH);
  }
@Test
  void testRequiredOptions() {
    ResolvedSchema schema1 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.TIMESTAMP(3))
        .build();
    final MockContext sourceContext1 = MockContext.getInstance(this.conf, schema1, "f2");

    // createDynamicTableSource doesn't call sanity check, will not throw exception
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext1));
    // miss pk and precombine key will throw exception when create sink
    assertThrows(HoodieValidationException.class, () -> new HoodieTableFactory().createDynamicTableSink(sourceContext1));

    // append mode does not throw
    this.conf.set(FlinkOptions.OPERATION, "insert");
    final MockContext sourceContext11 = MockContext.getInstance(this.conf, schema1, "f2");
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext11));
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSink(sourceContext11));
    //miss the pre combine key will be ok
    HoodieTableSink tableSink11 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sourceContext11);
    assertThat(tableSink11.getConf().getString(FlinkOptions.PRECOMBINE_FIELD), is(FlinkOptions.NO_PRE_COMBINE));
    this.conf.set(FlinkOptions.OPERATION, FlinkOptions.OPERATION.defaultValue());

    // a non-exists precombine key will throw exception
    ResolvedSchema schema2 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.TIMESTAMP(3))
        .build();
    this.conf.setString(FlinkOptions.PRECOMBINE_FIELD, "non_exist_field");
    final MockContext sourceContext2 = MockContext.getInstance(this.conf, schema2, "f2");
    // createDynamicTableSource doesn't call sanity check, will not throw exception
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext2));
    assertThrows(HoodieValidationException.class, () -> new HoodieTableFactory().createDynamicTableSink(sourceContext2));
    this.conf.setString(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.PRECOMBINE_FIELD.defaultValue());

    // given the pk but miss the pre combine key will be ok
    ResolvedSchema schema3 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.TIMESTAMP(3))
        .primaryKey("f0")
        .build();
    final MockContext sourceContext3 = MockContext.getInstance(this.conf, schema3, "f2");
    HoodieTableSource tableSource = (HoodieTableSource) new HoodieTableFactory().createDynamicTableSource(sourceContext3);
    HoodieTableSink tableSink = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sourceContext3);
    // the precombine field is overwritten
    assertThat(tableSink.getConf().getString(FlinkOptions.PRECOMBINE_FIELD), is(FlinkOptions.NO_PRE_COMBINE));
    // precombine field not specified, use the default payload clazz
    assertThat(tableSource.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(FlinkOptions.PAYLOAD_CLASS_NAME.defaultValue()));
    assertThat(tableSink.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(FlinkOptions.PAYLOAD_CLASS_NAME.defaultValue()));

    // append mode given the pk but miss the pre combine key will be ok
    this.conf.set(FlinkOptions.OPERATION, "insert");
    HoodieTableSink tableSink3 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sourceContext3);
    assertThat(tableSink3.getConf().getString(FlinkOptions.PRECOMBINE_FIELD), is(FlinkOptions.NO_PRE_COMBINE));
    this.conf.set(FlinkOptions.OPERATION, FlinkOptions.OPERATION.defaultValue());

    this.conf.setString(FlinkOptions.PAYLOAD_CLASS_NAME, DefaultHoodieRecordPayload.class.getName());
    final MockContext sourceContext4 = MockContext.getInstance(this.conf, schema3, "f2");

    // createDynamicTableSource doesn't call sanity check, will not throw exception
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext4));
    // given pk but miss the pre combine key with DefaultHoodieRecordPayload should throw
    assertThrows(HoodieValidationException.class, () -> new HoodieTableFactory().createDynamicTableSink(sourceContext4));
    this.conf.setString(FlinkOptions.PAYLOAD_CLASS_NAME, FlinkOptions.PAYLOAD_CLASS_NAME.defaultValue());

    // given pk and pre combine key will be ok
    ResolvedSchema schema4 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.TIMESTAMP(3))
        .field("ts", DataTypes.TIMESTAMP(3))
        .primaryKey("f0")
        .build();
    final MockContext sourceContext5 = MockContext.getInstance(this.conf, schema4, "f2");

    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext5));
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSink(sourceContext5));
    // precombine field specified(default ts), use DefaultHoodieRecordPayload as payload clazz
    HoodieTableSource tableSource5 = (HoodieTableSource) new HoodieTableFactory().createDynamicTableSource(sourceContext5);
    HoodieTableSink tableSink5 = (HoodieTableSink) new HoodieTableFactory().createDynamicTableSink(sourceContext5);
    assertThat(tableSource5.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(EventTimeAvroPayload.class.getName()));
    assertThat(tableSink5.getConf().getString(FlinkOptions.PAYLOAD_CLASS_NAME), is(EventTimeAvroPayload.class.getName()));

    // given pk and set pre combine key to no_precombine will be ok
    ResolvedSchema schema5 = SchemaBuilder.instance()
        .field("f0", DataTypes.INT().notNull())
        .field("f1", DataTypes.VARCHAR(20))
        .field("f2", DataTypes.TIMESTAMP(3))
        .field("ts", DataTypes.TIMESTAMP(3))
        .primaryKey("f0")
        .build();
    this.conf.setString(FlinkOptions.PRECOMBINE_FIELD, FlinkOptions.NO_PRE_COMBINE);
    final MockContext sourceContext6 = MockContext.getInstance(this.conf, schema5, "f2");

    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSource(sourceContext6));
    assertDoesNotThrow(() -> new HoodieTableFactory().createDynamicTableSink(sourceContext6));
  }