@Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        Configuration configuration = new Configuration();
        context.getCatalogTable().getOptions().forEach(configuration::setString);

        String inlongMetric = configuration.getOptional(INLONG_METRIC).orElse(null);
        String auditHostAndPorts = configuration.getOptional(INLONG_AUDIT).orElse(null);

        KuduTableInfo kuduTableInfo = getKuduTableInfo(context);

        return new KuduDynamicTableSource(
                kuduTableInfo,
                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