flink-connector-mongodb-cdc/src/main/java/com/ververica/cdc/connectors/mongodb/MongoDBSource.java
public DebeziumSourceFunction<T> build() {
            Properties props = new Properties();

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

            props.setProperty(
                    MongoSourceConfig.CONNECTION_URI_CONFIG, String.valueOf(buildConnectionUri()));

            props.setProperty(MongoSourceConfig.DATABASE_CONFIG, checkNotNull(database));
            props.setProperty(MongoSourceConfig.COLLECTION_CONFIG, checkNotNull(collection));

            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_VALUE_DEFAULT);

            if (pipeline != null) {
                props.setProperty(MongoSourceConfig.PIPELINE_CONFIG, pipeline);
            }

            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 (errorsLogEnable != null) {
                props.setProperty(
                        MongoSourceConfig.ERRORS_LOG_ENABLE_CONFIG,
                        String.valueOf(errorsLogEnable));
            }

            if (errorsTolerance != null) {
                props.setProperty(MongoSourceConfig.ERRORS_TOLERANCE_CONFIG, errorsTolerance);
            }

            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_DEFAULT);

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

            return new DebeziumSourceFunction<>(
                    deserializer, props, null, Validator.getDefaultValidator());
        }
/**
         * The properties of mongodb kafka connector.
         * https://docs.mongodb.com/kafka-connector/current/kafka-source
         */
The properties of mongodb kafka connector.
The properties of mongodb kafka connector .
public DebeziumSourceFunction<T> build() {
            Properties props = new Properties();
            props.setProperty(
                    "connector.class", MongoDBConnectorSourceConnector.class.getCanonicalName());
            props.setProperty("name", "mongodb_binlog_source");
            props.setProperty(
                    MongoSourceConfig.CONNECTION_URI_CONFIG, String.valueOf(buildConnectionUri()));
            props.setProperty(MongoSourceConfig.DATABASE_CONFIG, checkNotNull(database));
            props.setProperty(MongoSourceConfig.COLLECTION_CONFIG, checkNotNull(collection));
            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_VALUE_DEFAULT);
            if (pipeline != null) {
                props.setProperty(MongoSourceConfig.PIPELINE_CONFIG, pipeline);
            }
            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 (errorsLogEnable != null) {
                props.setProperty(
                        MongoSourceConfig.ERRORS_LOG_ENABLE_CONFIG,
                        String.valueOf(errorsLogEnable));
            }
            if (errorsTolerance != null) {
                props.setProperty(MongoSourceConfig.ERRORS_TOLERANCE_CONFIG, errorsTolerance);
            }
            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_DEFAULT);
            props.setProperty(
                    Heartbeat.HEARTBEAT_TOPICS_PREFIX.name(), HEARTBEAT_TOPIC_NAME_DEFAULT);
            return new DebeziumSourceFunction<>(
                    deserializer, props, null, Validator.getDefaultValidator());
        }
public DebeziumSourceFunction < T > build ( ) { Properties props = new Properties ( ) ; props . setProperty ( "connector.class" , MongoDBConnectorSourceConnector . class . getCanonicalName ( ) ) ; props . setProperty ( "name" , "mongodb_binlog_source" ) ; props . setProperty ( MongoSourceConfig . CONNECTION_URI_CONFIG , String . valueOf ( buildConnectionUri ( ) ) ) ; props . setProperty ( MongoSourceConfig . DATABASE_CONFIG , checkNotNull ( database ) ) ; props . setProperty ( MongoSourceConfig . COLLECTION_CONFIG , checkNotNull ( collection ) ) ; 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_VALUE_DEFAULT ) ; if ( pipeline != null ) { props . setProperty ( MongoSourceConfig . PIPELINE_CONFIG , pipeline ) ; } 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 ( errorsLogEnable != null ) { props . setProperty ( MongoSourceConfig . ERRORS_LOG_ENABLE_CONFIG , String . valueOf ( errorsLogEnable ) ) ; } if ( errorsTolerance != null ) { props . setProperty ( MongoSourceConfig . ERRORS_TOLERANCE_CONFIG , errorsTolerance ) ; } 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_DEFAULT ) ; props . setProperty ( Heartbeat . HEARTBEAT_TOPICS_PREFIX . name ( ) , HEARTBEAT_TOPIC_NAME_DEFAULT ) ; return new DebeziumSourceFunction < > ( deserializer , props , null , Validator . getDefaultValidator ( ) ) ; }
The properties of mongodb kafka connector.
The properties of mongodb kafka connector .
// Let DebeziumChangeFetcher recognize heartbeat record