private KuduTableInfo getKuduTableInfo(Context context) {
        TableSchema schema = context.getCatalogTable().getSchema();
        FactoryUtil.TableFactoryHelper helper =
                FactoryUtil.createTableFactoryHelper(this, context);

        ReadableConfig options = helper.getOptions();
        String master = options.getOptional(CONNECTOR_MASTERS).orElse(null);
        String tableName = options.getOptional(CONNECTOR_TABLE).orElse(null);

        SessionConfiguration.FlushMode flushMode = SessionConfiguration.FlushMode.AUTO_FLUSH_BACKGROUND;
        Optional<String> optional = options.getOptional(FLUSH_MODE);
        if (optional.isPresent()) {
            String flushModeConfig = optional.get();
            flushMode = SessionConfiguration.FlushMode.valueOf(flushModeConfig);
            checkNotNull(flushMode, "The flush mode must be one of " +
                    "AUTO_FLUSH_SYNC AUTO_FLUSH_BACKGROUND or MANUAL_FLUSH.");
        }

        return KuduTableInfo
                .builder()
                .masters(master)
                .tableName(tableName)
                .fieldNames(schema.getFieldNames())
                .dataTypes(
                        schema.getFieldDataTypes())
                .flushMode(flushMode)
                .build();
    }
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