@Override
  public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
    MixedTable table = MixedFormatUtils.loadMixedTable(tableLoader);

    return new DataStreamSinkProvider() {
      @Override
      public DataStreamSink<?> consumeDataStream(
          ProviderContext providerContext, DataStream<RowData> dataStream) {
        DataStreamSink<?> ds =
            FlinkSink.forRowData(dataStream)
                .context(providerContext)
                .table(table)
                .flinkSchema(flinkTable.getSchema())
                .tableLoader(tableLoader)
                .overwrite(overwrite)
                .build();
        UserGroupInformation.reset();
        LOG.info("ugi reset");
        return ds;
      }
    };
  }
Flink table api that generates sink operators.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedFormatDynamicSink.java