@Override
  public DynamicTableSource createDynamicTableSource(Context context) {
    Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions());
    StoragePath path = new StoragePath(conf.getOptional(FlinkOptions.PATH).orElseThrow(() ->
        new ValidationException("Option [path] should not be empty.")));
    setupTableOptions(conf.getString(FlinkOptions.PATH), conf);
    ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
    setupConfOptions(conf, context.getObjectIdentifier(), context.getCatalogTable(), schema);
    return new HoodieTableSource(
        SerializableSchema.create(schema),
        path,
        context.getCatalogTable().getPartitionKeys(),
        conf.getString(FlinkOptions.PARTITION_DEFAULT_NAME),
        conf);
  }
@Test
  void testSetupTimestampBasedKeyGenForSink() {
    this.conf.setString(FlinkOptions.RECORD_KEY_FIELD, "dummyField");
    // 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.TIMESTAMP(3))
        .field("ts", DataTypes.TIMESTAMP(3))
        .primaryKey("f0")
        .build();
    final MockContext sourceContext1 = MockContext.getInstance(this.conf, schema1, "ts");
    final HoodieTableSource tableSource1 = (HoodieTableSource) new HoodieTableFactory().createDynamicTableSource(sourceContext1);
    final Configuration conf1 = tableSource1.getConf();
    assertThat(conf1.get(FlinkOptions.RECORD_KEY_FIELD), is("f0"));
    assertThat(conf1.get(FlinkOptions.KEYGEN_CLASS_NAME), is(TimestampBasedAvroKeyGenerator.class.getName()));
    assertThat(conf1.getString(TIMESTAMP_TYPE_FIELD.key(), "dummy"),
        is("EPOCHMILLISECONDS"));
    assertThat(conf1.getString(TIMESTAMP_OUTPUT_DATE_FORMAT.key(), "dummy"),
        is(FlinkOptions.PARTITION_FORMAT_HOUR));
    assertThat(conf1.getString(TIMESTAMP_OUTPUT_TIMEZONE_FORMAT.key(), "dummy"),
        is("UTC"));
  }