@Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        TableSchema schema = context.getCatalogTable().getSchema();
        FactoryUtil.TableFactoryHelper helper =
                FactoryUtil.createTableFactoryHelper(this, context);

        Configuration configuration = new Configuration();
        context.getCatalogTable().getOptions().forEach(configuration::setString);

        ReadableConfig options = helper.getOptions();
        String inlongMetric = options.getOptional(INLONG_METRIC).orElse(null);
        String auditHostAndPorts = options.getOptional(INLONG_AUDIT).orElse(null);

        // Query the fields through the kudu client and select the fields in the tableSchema
        return new KuduDynamicTableSink(
                getKuduTableInfo(context),
                configuration,
                inlongMetric,
                auditHostAndPorts);
    }
Factory for creating configured instances of {@link KuduDynamicTableSink}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/table/KuduDynamicTableFactory.java