public DebeziumSourceFunction<T> build() {
            Properties props = new Properties();

            props.setProperty(
                    "connector.class", MongoDBConnectorSourceConnector.class.getCanonicalName());
            props.setProperty("name", "mongodb_cdc_source");

            props.setProperty(
                    MongoSourceConfig.CONNECTION_URI_CONFIG,
                    String.valueOf(
                            buildConnectionString(username, password, hosts, connectionOptions)));

            if (databaseList != null) {
                props.setProperty(DATABASE_INCLUDE_LIST, String.join(",", databaseList));
            }

            if (collectionList != null) {
                props.setProperty(COLLECTION_INCLUDE_LIST, String.join(",", collectionList));
            }

            if (updateLookup) {
                props.setProperty(
                        MongoSourceConfig.FULL_DOCUMENT_CONFIG, FULL_DOCUMENT_UPDATE_LOOKUP);
            }

            props.setProperty(
                    MongoSourceConfig.PUBLISH_FULL_DOCUMENT_ONLY_CONFIG,
                    String.valueOf(Boolean.FALSE));

            props.setProperty(MongoSourceConfig.OUTPUT_FORMAT_KEY_CONFIG, OUTPUT_FORMAT_SCHEMA);
            props.setProperty(MongoSourceConfig.OUTPUT_FORMAT_VALUE_CONFIG, OUTPUT_FORMAT_SCHEMA);
            props.setProperty(
                    MongoSourceConfig.OUTPUT_SCHEMA_INFER_VALUE_CONFIG,
                    String.valueOf(Boolean.FALSE));
            props.setProperty(MongoSourceConfig.OUTPUT_SCHEMA_VALUE_CONFIG, OUTPUT_SCHEMA);

            if (batchSize != null) {
                props.setProperty(MongoSourceConfig.BATCH_SIZE_CONFIG, String.valueOf(batchSize));
            }

            if (pollAwaitTimeMillis != null) {
                props.setProperty(
                        MongoSourceConfig.POLL_AWAIT_TIME_MS_CONFIG,
                        String.valueOf(pollAwaitTimeMillis));
            }

            if (pollMaxBatchSize != null) {
                props.setProperty(
                        MongoSourceConfig.POLL_MAX_BATCH_SIZE_CONFIG,
                        String.valueOf(pollMaxBatchSize));
            }

            if (copyExisting != null) {
                props.setProperty(
                        MongoSourceConfig.COPY_EXISTING_CONFIG, String.valueOf(copyExisting));
            }

            if (copyExistingMaxThreads != null) {
                props.setProperty(
                        MongoSourceConfig.COPY_EXISTING_MAX_THREADS_CONFIG,
                        String.valueOf(copyExistingMaxThreads));
            }

            if (copyExistingQueueSize != null) {
                props.setProperty(
                        MongoSourceConfig.COPY_EXISTING_QUEUE_SIZE_CONFIG,
                        String.valueOf(copyExistingQueueSize));
            }

            if (copyExistingPipeline != null) {
                props.setProperty(
                        MongoSourceConfig.COPY_EXISTING_PIPELINE_CONFIG, copyExistingPipeline);
            }

            if (heartbeatIntervalMillis != null) {
                props.setProperty(
                        MongoSourceConfig.HEARTBEAT_INTERVAL_MS_CONFIG,
                        String.valueOf(heartbeatIntervalMillis));
            }

            props.setProperty(MongoSourceConfig.HEARTBEAT_TOPIC_NAME_CONFIG, HEARTBEAT_TOPIC_NAME);

            // Let DebeziumChangeFetcher recognize heartbeat record
            props.setProperty(Heartbeat.HEARTBEAT_TOPICS_PREFIX.name(), HEARTBEAT_TOPIC_NAME);

            props.setProperty(
                    MongoSourceConfig.ERRORS_LOG_ENABLE_CONFIG, String.valueOf(Boolean.TRUE));
            props.setProperty(
                    MongoSourceConfig.ERRORS_TOLERANCE_CONFIG, ErrorTolerance.NONE.value());

            return new DebeziumSourceFunction<>(
                    deserializer, props, null, Validator.getDefaultValidator(), metricOption);
        }
The properties of mongodb kafka connector.
https://docs.mongodb.com/kafka-connector/current/kafka-source
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/mongodb/MongoDBSource.java