@Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        final FactoryUtil.TableFactoryHelper helper =
                FactoryUtil.createTableFactoryHelper(this, context);
        helper.validate();

        final ReadableConfig config = helper.getOptions();

        final String hosts = config.get(HOSTS);
        final String connectionOptions = config.getOptional(CONNECTION_OPTIONS).orElse(null);

        final String username = config.getOptional(USERNAME).orElse(null);
        final String password = config.getOptional(PASSWORD).orElse(null);

        final String database = config.getOptional(DATABASE).orElse(null);
        final String collection = config.getOptional(COLLECTION).orElse(null);

        Integer batchSize = config.get(BATCH_SIZE);
        final Integer pollMaxBatchSize = config.get(POLL_MAX_BATCH_SIZE);
        final Integer pollAwaitTimeMillis = config.get(POLL_AWAIT_TIME_MILLIS);

        final Integer heartbeatIntervalMillis = config.get(HEARTBEAT_INTERVAL_MILLIS);

        final Boolean copyExisting = config.get(COPY_EXISTING);
        final Integer copyExistingQueueSize = config.getOptional(COPY_EXISTING_QUEUE_SIZE).orElse(null);

        final String zoneId = context.getConfiguration().get(TableConfigOptions.LOCAL_TIME_ZONE);
        final ZoneId localTimeZone =
                TableConfigOptions.LOCAL_TIME_ZONE.defaultValue().equals(zoneId)
                        ? ZoneId.systemDefault()
                        : ZoneId.of(zoneId);

        boolean enableParallelRead = config.get(SCAN_INCREMENTAL_SNAPSHOT_ENABLED);

        int splitSizeMB = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE_MB);
        int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE);

        final String inlongMetric = config.getOptional(INLONG_METRIC).orElse(null);
        final String inlongAudit = config.get(INLONG_AUDIT);
        final Boolean sourceMultipleEnable = config.get(SOURCE_MULTIPLE_ENABLE);
        ResolvedSchema physicalSchema = context.getCatalogTable().getResolvedSchema();
        if (!sourceMultipleEnable) {
            checkArgument(physicalSchema.getPrimaryKey().isPresent(), "Primary key must be present");
            checkPrimaryKey(physicalSchema.getPrimaryKey().get(), "Primary key must be _id field");
        }
        final String rowKindFiltered = config.get(ROW_KINDS_FILTERED).isEmpty()
                ? ROW_KINDS_FILTERED.defaultValue()
                : config.get(ROW_KINDS_FILTERED);
        Boolean changelogNormalizeEnabled = config.get(CHANGELOG_NORMALIZE_ENABLED);

        return new MongoDBTableSource(
                physicalSchema,
                hosts,
                username,
                password,
                database,
                collection,
                connectionOptions,
                copyExisting,
                copyExistingQueueSize,
                batchSize,
                pollMaxBatchSize,
                pollAwaitTimeMillis,
                heartbeatIntervalMillis,
                localTimeZone,
                enableParallelRead,
                splitMetaGroupSize,
                splitSizeMB,
                inlongMetric,
                inlongAudit,
                rowKindFiltered,
                sourceMultipleEnable,
                changelogNormalizeEnabled);
    }
Factory for creating configured instance of {@link MongoDBTableSource}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/table/MongoDBTableSourceFactory.java