MongoDB Source Connector - AutoMQ Documentation

Overview

The MongoDB Source Connector uses MongoDB Change Streams to read changes from a database, a collection, or an entire deployment and write the events to Kafka topics. It is suitable for data pipelines such as data synchronization, stream processing, search indexing, and auditing. By default, the MongoDB namespace database.collection maps to a Kafka topic with the same name, and the message value contains the complete change stream event. The Connector can capture only changes that occur after startup, or it can first copy existing documents and then continue with subsequent changes. Copied documents are wrapped as Change Stream events shaped like insert operations and pass through the same topic, key, and value mapping process as real-time events.

Prerequisites

The MongoDB deployment must support Change Streams. The connection account needs the permissions required to read change streams for the capture scope, and it also needs permission to read the target collections when copying existing data.
When using MongoDB capabilities such as pre-change documents, expanded events, or the Stable API, the server version and database settings must meet the requirements of the corresponding feature.

License

Uses Apache License 2.0.

Quick Start

Prepare a Connect Cluster, Kafka, and a MongoDB deployment that supports Change Streams, and confirm network connectivity and access permissions. For preparation and management, see Manage connectors.

connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=<database-name>
collection=<collection-name>

Replace the MongoDB connection information, database name, and collection name. This configuration starts capturing from the current change stream position and writes events to the <database-name>.<collection-name> topic by default. Use a Config Provider or protected configuration management for connection credentials.

Configuration

Connector and Tasks

`connector.class`

Specifies the MongoDB Source Connector implementation class.

Type: string
Default: None
Importance: High
Valid Values / Notes: Use com.mongodb.kafka.connect.MongoSourceConnector.
Required: Yes

`tasks.max`

Sets the maximum number of Tasks that Connect may create.

Type: int
Default: 1
Importance: High
Valid Values / Notes: Must be at least 1. This Connector always creates only one Source Task, so a larger value does not increase Task parallelism.

MongoDB Connection and Capture Scope

`connection.uri`

The MongoDB connection string, which specifies hosts, authentication, the Replica Set, TLS, and other driver options.

Type: password
Default: mongodb://localhost:27017,localhost:27018,localhost:27019
Importance: High
Valid Values / Notes: Must be a valid MongoDB URI. The default is generally suitable only for local development. Do not store credentials in plaintext configuration.

`database`

Restricts monitoring to a database.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: When empty, monitors the entire MongoDB deployment. Setting a database without setting a collection monitors the collections in that database.

`collection`

Restricts monitoring to a single collection.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: This is not a regular expression or a list. When set, database must also be set.

`offset.partition.name`

Overrides the name the Connector uses to identify the Source Offset partition.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: When empty, the name is generated from the connection hosts, database, and collection. Changing this value starts using a new Offset identity; the existing Resume Token is not migrated automatically.

MongoDB Stable API

`server.api.version`

Enables the specified version of the MongoDB Stable API.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: An empty value disables it. The available value is 1, which requires MongoDB 5.0 or later.

`server.api.deprecation.errors`

Enables Stable API deprecation errors.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only when server.api.version is not empty.

`server.api.strict`

Enables Stable API strict mode.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only when server.api.version is not empty.

TLS Certificates

`connection.ssl.truststore`

Specifies the Truststore path local to the Connect Worker.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: When not empty, it is loaded using the JVM default KeyStore type. TLS must still be enabled through the connection URI or deployment settings.

`connection.ssl.truststorePassword`

Specifies the Truststore password.

Type: password
Default: Empty string
Importance: Medium
Valid Values / Notes: Used only when connection.ssl.truststore is not empty.

`connection.ssl.keystore`

Specifies the Keystore path local to the Connect Worker for providing a client certificate and private key.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: When not empty, it is loaded using the JVM default KeyStore type.

`connection.ssl.keystorePassword`

Specifies the password for the Keystore and its private key.

Type: password
Default: Empty string
Importance: Medium
Valid Values / Notes: Used only when connection.ssl.keystore is not empty. The same password is used to load the KeyStore and initialize the private key.

Change Stream

`pipeline`

Sets an aggregation pipeline for the Change Stream.

Type: string
Default: []
Importance: Medium
Valid Values / Notes: Must be an array of JSON documents. The pipeline must preserve the _id, ns, documentKey, or fullDocument fields required for topic, Offset, key, and value mapping. For efficient filtering while copying existing data, prefer startup.mode.copy.existing.pipeline.

`batch.size`

Sets the batch size hint for the MongoDB Change Stream Cursor.

Type: int
Default: 0
Importance: Medium
Valid Values / Notes: Must be at least 0. 0 retains driver and server behavior and is different from poll.max.batch.size.

`publish.full.document.only`

Publishes only the change event’s fullDocument as the message value.

Type: boolean
Default: false
Importance: High
Valid Values / Notes: When set to true, forces updateLookup. Events without fullDocument are filtered unless delete Tombstones are also enabled.

`publish.full.document.only.tombstone.on.delete`

In full-document-only mode, sends a null-valued Tombstone for events that do not contain a document-valued fullDocument; delete events are the most common case.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only when publish.full.document.only=true. The implementation checks for a document-valued fullDocument; it does not only check whether the event’s operationType is delete.

`change.stream.document.key.as.key`

Controls whether non-Schema output uses the Change Stream documentKey as the Kafka message key.

Type: boolean
Default: true
Importance: Medium
Valid Values / Notes: When set to false, or when the event has no documentKey, the key uses the Resume Token document. This setting is ignored when output.format.key=schema.

`change.stream.full.document.before.change`

Sets how the Change Stream returns the document from before the change.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Available values are default, off, whenAvailable, required, or an empty string. Requires MongoDB 6.0 or later and Pre-images enabled for the collection. Does not affect records copied from existing data.

`change.stream.full.document`

Sets how the Change Stream returns the full document.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Available values are default, updateLookup, whenAvailable, required, or an empty string. publish.full.document.only=true forces updateLookup.

`change.stream.show.expanded.events`

Controls whether expanded Change Stream events are requested.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Expanded DDL events require MongoDB 6.0 or later, and disambiguated update paths require MongoDB 6.1 or later.

`collation`

Sets the Collation for the Change Stream.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: An empty value leaves it unset; otherwise, it must be a valid MongoDB Collation JSON document. This setting does not apply to aggregation for copying existing data.

`poll.max.batch.size`

Limits the maximum number of records returned by a single poll.

Type: int
Default: 1000
Importance: Low
Valid Values / Notes: Must be at least 1. It limits only the number of records, not the total bytes in a batch.

`poll.await.time.ms`

Sets the maximum time that the Change Stream waits for new events.

Type: long
Default: 5000
Importance: Low
Valid Values / Notes: Must be at least 1, in milliseconds. This value is not the Kafka Connect Poll Timeout.

Topic Mapping

`topic.mapper`

Specifies the implementation class that maps MongoDB namespaces to Kafka topics.

Type: string
Default: com.mongodb.kafka.connect.source.topic.mapping.DefaultTopicMapper
Importance: High
Valid Values / Notes: Must be a loadable fully qualified Java class name that implements TopicMapper. A custom implementation may read additional properties that it defines.

`topic.separator`

Specifies the separator between parts of a default topic name.

Type: string
Default: .
Importance: Low
Valid Values / Notes: Used to join the prefix, database, collection, and suffix. The Connector does not validate whether the final topic name is valid.

`topic.prefix`

Adds a prefix to all topics produced by the default mapping.

Type: string
Default: Empty string
Importance: Low
Valid Values / Notes: A non-empty prefix is followed by topic.separator.

`topic.suffix`

Adds a suffix to all topics produced by the default mapping.

Type: string
Default: Empty string
Importance: Low
Valid Values / Notes: A non-empty suffix is preceded by topic.separator.

`topic.namespace.map`

Uses a JSON object to map MongoDB namespaces to specified topics.

Type: string
Default: Empty string
Importance: High
Valid Values / Notes: Supports full namespaces, databases, /regular expression/ mappings matched in declaration order, and * mappings. Prefixes and suffixes are still applied to mapping results. Templates may use the db, sep, coll, sep_coll, coll_sep, and sep_coll_sep variables. You must prevent invalid names and topic conflicts.

Output Format and Schema

`output.format.key`

Sets the output format of message keys.

Type: string
Default: json
Importance: High
Valid Values / Notes: Available values are json, bson, and schema, case-insensitive. The format should be compatible with key.converter.

`output.format.value`

Sets the output format of message values.

Type: string
Default: json
Importance: High
Valid Values / Notes: Available values are json, bson, and schema, case-insensitive. The format should be compatible with value.converter.

`output.schema.infer.value`

Automatically infers a Connect Schema for each message value.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only when output.format.value=schema. Changes in document structure may cause multiple Schemas to appear in the same topic.

`output.schema.key`

Defines the Avro Schema used for Schema-format message keys.

Type: string
Default: { "type": "record", "name": "keySchema", "fields" : [{"name": "_id", "type": "string"}]}
Importance: High
Valid Values / Notes: Must be a valid Connector Avro Schema compatible with conversion of the complete Change Stream document. Read only when output.format.key=schema.

`output.schema.value`

Defines the Avro Schema used for Schema-format message values.

Type: string
Default: Built-in ChangeStream Avro Schema
Importance: High
Valid Values / Notes: Must be a valid Connector Avro Schema compatible with the message value. Ignored when output.schema.infer.value=true or the output format is not schema.

`output.json.formatter`

Specifies the BSON-to-JSON formatting implementation class.

Type: string
Default: com.mongodb.kafka.connect.source.json.formatter.DefaultJson
Importance: Medium
Valid Values / Notes: Must be a loadable fully qualified Java class name that implements JsonWriterSettingsProvider. It is used for JSON output and Schema conversion and ignored for raw BSON output.

`key.converter`

Overrides the Worker-level message key Converter.

Type: class
Default: null
Importance: Low
Valid Values / Notes: null inherits the Worker configuration. The selected Converter must be compatible with the String, BSON Bytes, or Schemaful Connect Data produced by output.format.key.

`value.converter`

Overrides the Worker-level message value Converter.

Type: class
Default: null
Importance: Low
Valid Values / Notes: null inherits the Worker configuration. The selected Converter must be compatible with the String, BSON Bytes, or Schemaful Connect Data produced by output.format.value.

Startup Position and Existing Data Copy

`startup.mode`

Sets the startup behavior when no usable Source Offset is available.

Type: string
Default: Empty string, treated as latest
Importance: Medium
Valid Values / Notes: Available values are latest, timestamp, copy_existing, or an empty string. An existing Resume Token takes precedence over this setting. An explicit non-empty value takes precedence over the deprecated copy.existing.

`startup.mode.timestamp.start.at.operation.time`

Sets the starting operation time for Timestamp startup mode.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Applies only when startup.mode=timestamp and no usable Offset is available. Supports decimal Epoch seconds, an ISO-8601 Instant with second precision, or a Canonical Extended JSON BSON Timestamp.

`startup.mode.copy.existing.max.threads`

Sets the number of read threads for copying existing data.

Type: int
Default: Runtime.getRuntime().availableProcessors() (number of processors available to the runtime)
Importance: Medium
Valid Values / Notes: Must be at least 1 and is used only when startup.mode=copy_existing. The actual thread count does not exceed the number of selected namespaces. When explicitly set, it overrides the deprecated copy.existing.max.threads.

`startup.mode.copy.existing.queue.size`

Sets the capacity of the in-memory queue between existing-data copy threads and the Source Task.

Type: int
Default: 16000
Importance: Medium
Valid Values / Notes: Must be at least 1 and is used only when startup.mode=copy_existing. When explicitly set, it overrides the deprecated copy.existing.queue.size.

`startup.mode.copy.existing.pipeline`

Sets an aggregation pipeline used only to read existing documents.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Must be empty or an array of JSON documents and is used only when startup.mode=copy_existing. It runs before synthetic change events are created and before the general pipeline. When explicitly set, it overrides the deprecated copy.existing.pipeline.

`startup.mode.copy.existing.namespace.regex`

Uses a regular expression to filter existing-data copy namespaces discovered at startup.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Matches the full database.collection using Java regular expression substring matching semantics and is used only when startup.mode=copy_existing. When explicitly set, it overrides the deprecated copy.existing.namespace.regex.

`startup.mode.copy.existing.allow.disk.use`

Controls whether MongoDB may use disk for aggregation while copying existing data.

Type: boolean
Default: true
Importance: Medium
Valid Values / Notes: Used only when startup.mode=copy_existing. When explicitly set, it overrides the deprecated copy.existing.allow.disk.use.

`copy.existing`

Enables copying existing data with the legacy switch.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: Applies only when startup.mode is empty.
Deprecated: Yes
Replacement: startup.mode=copy_existing

`copy.existing.max.threads`

Sets the legacy thread count for copying existing data.

Type: int
Default: Runtime.getRuntime().availableProcessors() (number of processors available to the runtime)
Importance: Medium
Valid Values / Notes: Must be at least 1. It is used as a fallback only when the replacement is not explicitly set.
Deprecated: Yes
Replacement: startup.mode.copy.existing.max.threads

`copy.existing.queue.size`

Sets the legacy in-memory queue capacity for copying existing data.

Type: int
Default: 16000
Importance: Medium
Valid Values / Notes: Must be at least 1. It is used as a fallback only when the replacement is not explicitly set.
Deprecated: Yes
Replacement: startup.mode.copy.existing.queue.size

`copy.existing.pipeline`

Sets the legacy aggregation pipeline for copying existing data.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Must be empty or an array of JSON documents. It is used as a fallback only when the replacement is not explicitly set.
Deprecated: Yes
Replacement: startup.mode.copy.existing.pipeline

`copy.existing.namespace.regex`

Sets the legacy namespace regular expression for copying existing data.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Must be a valid Java regular expression. It is used as a fallback only when the replacement is not explicitly set.
Deprecated: Yes
Replacement: startup.mode.copy.existing.namespace.regex

`copy.existing.allow.disk.use`

Sets whether legacy existing-data copy aggregation may use disk.

Type: boolean
Default: true
Importance: Medium
Valid Values / Notes: It is used as a fallback only when the replacement is not explicitly set.
Deprecated: Yes
Replacement: startup.mode.copy.existing.allow.disk.use

Error Handling and Heartbeats

`errors.tolerance`

Sets the error tolerance mode for the Kafka Connect framework and the MongoDB Connector.

Type: string
Default: none
Importance: Medium
Valid Values / Notes: Available values are none and all, case-insensitive. all can skip conversion failures and allow some Change Stream recovery paths to rebuild without a Resume Token. mongo.errors.tolerance overrides only the Connector’s reading of this value.

`mongo.errors.tolerance`

Overrides the error tolerance mode only within the MongoDB Connector.

Type: string
Default: none
Importance: Medium
Valid Values / Notes: Available values are none and all. Whenever explicitly set, it takes precedence over the Connector’s reading of errors.tolerance, but it does not replace the Kafka Connect framework setting with the same name.

`errors.log.enable`

Controls whether the Kafka Connect framework and the MongoDB Connector log tolerated errors.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: The Connector still logs conversion errors when the tolerance mode is none. mongo.errors.log.enable can override only the Connector’s behavior.

`mongo.errors.log.enable`

Overrides only whether the MongoDB Connector logs tolerated conversion errors.

Type: boolean
Default: false
Importance: Medium
Valid Values / Notes: When explicitly set, it takes precedence over the Connector’s reading of errors.log.enable and does not change the Kafka Connect framework logging setting.

`errors.deadletterqueue.topic.name`

Sets the DLQ topic to which records with MongoDB conversion failures are written.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Used only when the Connector’s effective tolerance mode is all. The Connector does not create or validate the topic. mongo.errors.deadletterqueue.topic.name can override the Connector’s reading of this value.

`mongo.errors.deadletterqueue.topic.name`

Overrides only the DLQ topic used by the MongoDB Connector for records with conversion failures.

Type: string
Default: Empty string
Importance: Medium
Valid Values / Notes: Requires the Connector’s effective error tolerance mode to be all. It affects only Connector conversion error handling.

`heartbeat.interval.ms`

Sets the minimum interval for generating heartbeat records when there are no business records.

Type: long
Default: 0
Importance: Medium
Valid Values / Notes: Must be at least 0, in milliseconds. 0 disables heartbeats. A heartbeat is generated only when a new Post-batch Resume Token appears and the current Poll contains no business records.

`heartbeat.topic.name`

Specifies the Kafka topic to which heartbeat records are written.

Type: string
Default: __mongodb_heartbeats
Importance: Medium
Valid Values / Notes: Must be a non-empty string and is used only when heartbeat.interval.ms is greater than 0 and a heartbeat is generated. The topic must already meet naming, permission, and availability requirements.

Custom Credentials

`mongo.custom.auth.mechanism.enable`

Enables a custom MongoDB credential provider.

Type: string
Default: None
Importance: Not declared
Valid Values / Notes: Only the string value true, case-insensitive, enables it. This property is not registered in ConfigDef, so it does not appear in ConfigDef metadata and does not receive ConfigDef type validation.

`mongo.custom.auth.mechanism.providerClass`

Specifies the custom MongoDB credential provider implementation class.

Type: string
Default: None
Importance: Not declared
Valid Values / Notes: Required when custom authentication is enabled. The class must be loadable, implement CustomCredentialProvider, have an accessible no-argument constructor, and be available with its dependencies on every Worker where the Task may run.

Best Practices

Initially Load an Existing Collection and Continue Capturing Changes

Applicable scenario: The target collection already contains existing documents, and you need to establish a Kafka data baseline before continuously receiving new changes produced during and after the copy. This mode is suitable for initial onboarding, not as an initialization process to repeat on every restart. Use the following complete Connector configuration based on the Quick Start configuration:

connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=<database-name>
collection=<collection-name>
startup.mode=copy_existing

Key considerations: At startup, the Connector copies existing documents in the selected scope and then continues with the Change Stream. Restarting after an interrupted copy repeats the entire copy, so consumers should handle duplicates by using stable business keys or another method. The copy is not a consistent snapshot across multiple collections.

Assign Stable Topic Names to Business Namespaces

Applicable scenario: MongoDB database or collection names are unsuitable as downstream topic contracts, or existing topic names must remain unchanged during a migration. Explicit mapping decouples source naming from the Kafka consumer interface. Use the following Connector configuration to route one namespace to a fixed topic:

connector.class=com.mongodb.kafka.connect.MongoSourceConnector
connection.uri=mongodb://<username>:<password>@<mongodb-host>:27017/?replicaSet=<replica-set>
database=inventory
collection=orders
topic.namespace.map={"inventory.orders":"commerce.orders"}

Key considerations: Mapping occurs before prefix and suffix processing. Changing the mapping does not change the Source Offset identity, but you must confirm in advance the target topic’s naming, permissions, partitioning strategy, and potential conflicts with other mapping results.

Monitoring

What to Monitor

Monitor Kafka Connect health, Connector and Task status, throughput, latency, Offset commits, errors, retries, and Worker JVM signals; monitor DLQ activity only when the corresponding error handling is enabled.

Import the Grafana Dashboard

Ensure that Connect metrics are available in a Grafana data source and that the collected labels meet the dashboard filter requirements; download the Kafka Connect Dashboard, import the JSON into Grafana, and select the corresponding data source.

Limitations

Each Connector instance creates only one Source Task. Increasing tasks.max cannot horizontally scale Change Stream consumption.
Restarting during copy_existing repeats the copy from the beginning, so existing records already written to Kafka may be duplicated. The process is also not a consistent snapshot across collections.
Copying multiple namespaces may interleave records, topic mapping may split data streams, and the Connector does not specify the Kafka Partition. Therefore, it does not guarantee global ordering across namespaces, topics, or Partitions.
publish.full.document.only=true filters events without fullDocument unless the Tombstone option is also enabled to generate null-valued records.
When error tolerance is set to all without a configured DLQ, records with conversion failures are discarded, and subsequent Offsets may permanently advance past those events.
When a Resume Token is invalid or its history has been lost, a tolerant recovery path may rebuild the Change Stream without a Token and cannot recover a data interval that has already left the Oplog.
The Connector does not provide end-to-end Exactly-once, duplicate-free, or lossless guarantees from MongoDB to Kafka. Offset commit windows, copying existing data, and error tolerance can all produce duplicates or data gaps.

FAQ

No New Messages Arrive After the Connector Starts

Check whether the MongoDB deployment supports Change Streams, whether the connection account has permissions for the selected scope, and whether database and collection point to the location receiving writes. Confirm that the default <database>.<collection> topic or the target topic from topic.namespace.map is correct. If you use pipeline, also confirm that it does not filter out all events or remove required fields such as _id and ns.

Update Events Do Not Contain the Latest Full Document

By default, a Change Stream event does not necessarily contain the complete document after an update. Set change.stream.full.document=updateLookup when the complete document is required. To publish only complete documents, set publish.full.document.only=true, but also evaluate the filtering of events without fullDocument and whether delete events require Tombstones.

Capture Starts from an Unexpected Position After a Restart

The Connector prioritizes the committed Resume Token and applies startup.mode only when no usable Offset is available. Check whether the connection hosts, database, collection, or offset.partition.name changed, because these changes may create a new Offset identity. Restore the previous configuration, confirm that the original Offset still exists, and then restart the Connector.

Duplicate Records Appear After Copying Existing Data

Documents read during an existing-data copy are published as insert events. Restarting after an interrupted copy repeats the entire copy, and concurrent updates may also appear as subsequent Change Stream events. Consumers should implement idempotent processing based on stable document keys and determine before restarting whether to continue using the original Offset or establish a new data baseline.

Conversion Errors Do Not Enter the DLQ

Confirm that the effective Connector error tolerance mode is all, that the DLQ topic exists and the Worker has write permission, and whether any mongo.errors.* overrides are set. Filtering behavior such as an empty topic mapping or a missing fullDocument in full-document-only mode is not a conversion failure and is not written to the DLQ.

​Overview

​Prerequisites

​License

​Quick Start

​Configuration

​Connector and Tasks

​connector.class

​tasks.max

​MongoDB Connection and Capture Scope

​connection.uri

​database

​collection

​offset.partition.name

​MongoDB Stable API

​server.api.version

​server.api.deprecation.errors

​server.api.strict

​TLS Certificates

​connection.ssl.truststore

​connection.ssl.truststorePassword

​connection.ssl.keystore

​connection.ssl.keystorePassword

​Change Stream

​pipeline

​batch.size

​publish.full.document.only

​publish.full.document.only.tombstone.on.delete

​change.stream.document.key.as.key

​change.stream.full.document.before.change

​change.stream.full.document

​change.stream.show.expanded.events

​collation

​poll.max.batch.size

​poll.await.time.ms

​Topic Mapping

​topic.mapper

​topic.separator

​topic.prefix

​topic.suffix

​topic.namespace.map

​Output Format and Schema

​output.format.key

​output.format.value

​output.schema.infer.value

​output.schema.key

​output.schema.value

​output.json.formatter

​key.converter

​value.converter

​Startup Position and Existing Data Copy

​startup.mode

​startup.mode.timestamp.start.at.operation.time

​startup.mode.copy.existing.max.threads

​startup.mode.copy.existing.queue.size

​startup.mode.copy.existing.pipeline

​startup.mode.copy.existing.namespace.regex

​startup.mode.copy.existing.allow.disk.use

​copy.existing

​copy.existing.max.threads

​copy.existing.queue.size

​copy.existing.pipeline

​copy.existing.namespace.regex

​copy.existing.allow.disk.use

​Error Handling and Heartbeats

​errors.tolerance

​mongo.errors.tolerance

​errors.log.enable

​mongo.errors.log.enable

​errors.deadletterqueue.topic.name

​mongo.errors.deadletterqueue.topic.name

​heartbeat.interval.ms

​heartbeat.topic.name

​Custom Credentials

​mongo.custom.auth.mechanism.enable

​mongo.custom.auth.mechanism.providerClass

​Best Practices

​Initially Load an Existing Collection and Continue Capturing Changes

​Assign Stable Topic Names to Business Namespaces

​Monitoring

​What to Monitor