Skip to main content

​
Overview

The Debezium MongoDB CDC Source Connector reads document changes from MongoDB deployments that support change streams and publishes document create, update, and delete events to Kafka topics. Topic names consist of the logical prefix, database, and collection. The initial snapshot first publishes existing documents in the matching collections, after which the connector continues publishing change stream events. Snapshot records use the READ operation, while inserts, updates, and deletes correspond to the CREATE, UPDATE, and DELETE operations; delete events also produce tombstone records by default. It is suitable for sending MongoDB business data to data lakes, search systems, analytics systems, or downstream services in real time, and for CDC pipelines that initialize existing data before continuing with ongoing synchronization. Source recovery depends on the offsets persisted by Kafka Connect and on MongoDB still retaining the corresponding change stream history; consumers should handle possible duplicate events using stable keys or source positions.

​
Prerequisites

  • MongoDB must be deployed as a replica set or another supported cluster deployment that provides Change Streams; standalone MongoDB cannot be used as the change stream source for this Connector.
  • The account used to connect to MongoDB must have permission to read the target databases and collections. When signals or incremental snapshots are enabled, it must also be able to read the signal collection. Signals are written by the external client that issues them, and that client needs the corresponding write permission.
  • When TLS is used, the Kafka Connect Worker must be able to read the corresponding truststore, keystore, and passwords, and the hostname on the MongoDB server certificate must pass validation.

​
License

Uses Apache License 2.0.

​
Quick Start

Prepare an accessible Kafka Connect cluster, Kafka cluster, and MongoDB change stream source, confirm that the Worker can access MongoDB and the Connector plugin directory, and prepare an account authorized to read the source. To create and manage a Connector, see AutoMQ’s Manage Connectors.
Replace <database-host>, <database-port>, <replica-set-name>, <mongodb-user>, <mongodb-password>, and <logical-topic-prefix> with actual values. With the default snapshot.mode=initial, the Connector first snapshots matching source data when no usable prior offset exists, and then continues reading the change stream; the default MongoDB authentication source is admin. The Converter format should match the consumer contract.

​
Configuration

​
Connection and Authentication

​
topic.prefix

Provides a unique logical prefix for the MongoDB topics generated by the Connector.
  • Type: STRING
  • Default: No fixed default
  • Importance: High
  • Required: Yes
  • Valid values / notes: May contain only letters, digits, hyphens, periods, and underscores; different MongoDB Source instances in the same Kafka cluster should use different prefixes.

​
mongodb.connection.string

MongoDB connection string.
  • Type: STRING
  • Default: No fixed default
  • Importance: High
  • Required: Yes
  • Valid values / notes: Must be a valid MongoDB ConnectionString; the Connector establishes a live connection to validate the configuration. Replica set connections should provide replica set information in the connection string.

​
mongodb.user

MongoDB username.
  • Type: STRING
  • Default: No fixed default
  • Importance: High
  • Required: No
  • Valid values / notes: May be omitted when credentials are embedded in the connection string or when a custom authentication provider is used.

​
mongodb.password

MongoDB password.
  • Type: PASSWORD
  • Default: No fixed default
  • Importance: High
  • Required: No
  • Valid values / notes: Can be used together with mongodb.user; do not put the actual password in logs, documentation examples, or the repository.

​
mongodb.authsource

Authentication database that stores the MongoDB user credentials.
  • Type: STRING
  • Default: admin
  • Importance: Medium
  • Valid values / notes: Set this to the actual authentication database when the user was not created in the admin database.

​
mongodb.ssl.enabled

Whether to enable TLS/SSL for the MongoDB connection.
  • Type: BOOLEAN
  • Default: false
  • Importance: Medium
  • Valid values / notes: When enabled, also prepare the trust configuration required to validate the server certificate.

​
mongodb.ssl.invalid.hostname.allowed

Whether to allow hostname validation failures for the MongoDB server certificate.
  • Type: BOOLEAN
  • Default: false
  • Importance: Medium
  • Valid values / notes: true disables hostname protection and introduces man-in-the-middle attack risk; use only in controlled diagnostic environments.

​
mongodb.ssl.keystore

Path to the client keystore used for mutual TLS authentication.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Required: No
  • Valid values / notes: Set only when MongoDB requires a client certificate.

​
mongodb.ssl.keystore.password

Client keystore password.
  • Type: PASSWORD
  • Default: No fixed default
  • Importance: Medium
  • Required: No
  • Valid values / notes: Must be set when the keystore is password-protected.

​
mongodb.ssl.keystore.type

Client keystore type.
  • Type: STRING
  • Default: PKCS12
  • Importance: Medium
  • Valid values / notes: Takes effect only when mongodb.ssl.keystore is set and should match the actual keystore format.

​
mongodb.ssl.truststore

Path to the truststore used to validate the MongoDB server certificate.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Required: No
  • Valid values / notes: Set this when the Worker does not use the default truststore or when a dedicated CA is required.

​
mongodb.ssl.truststore.password

Truststore password.
  • Type: PASSWORD
  • Default: No fixed default
  • Importance: Medium
  • Required: No
  • Valid values / notes: Set this when the truststore must be unlocked or its password verified.

​
mongodb.ssl.truststore.type

Truststore type.
  • Type: STRING
  • Default: PKCS12
  • Importance: Medium
  • Valid values / notes: Takes effect only when mongodb.ssl.truststore is set and should match the actual truststore format.

​
mongodb.connect.timeout.ms

Timeout for establishing a MongoDB connection, in milliseconds.
  • Type: INT
  • Default: 10000
  • Importance: Low
  • Valid values / notes: Adjust according to network latency and deployment location; a value that is too small can cause startup or reconnect failures.

​
mongodb.server.selection.timeout.ms

Timeout for the MongoDB driver to select an available server, in milliseconds.
  • Type: INT
  • Default: 30000
  • Importance: Low
  • Valid values / notes: May be increased appropriately during server failover or in cross-network deployments.

​
mongodb.socket.timeout.ms

Timeout for MongoDB socket operations, in milliseconds.
  • Type: INT
  • Default: 0
  • Importance: Low
  • Valid values / notes: 0 means that no socket timeout is set.

​
mongodb.heartbeat.frequency.ms

MongoDB driver cluster-monitoring heartbeat frequency, in milliseconds.
  • Type: INT
  • Default: 10000
  • Importance: Low
  • Valid values / notes: Affects how quickly the driver detects changes in cluster status.

​
mongodb.poll.interval.ms

Interval for polling changes to MongoDB replica set members, in milliseconds.
  • Type: LONG
  • Default: 30000
  • Importance: Medium
  • Valid values / notes: Must be positive; shorten it when topology changes must be detected faster, but this increases polling overhead.

​
connection.validation.timeout.ms

Timeout for validating the MongoDB connection during Connector configuration validation, in milliseconds.
  • Type: LONG
  • Default: 60000
  • Importance: Low
  • Valid values / notes: Must be positive; it controls connection validation only and does not replace runtime connection timeout settings.

​
Source Scope and Fields

​
database.include.list

Specifies the databases to capture.
  • Type: LIST
  • Default: No fixed default
  • Importance: High
  • Valid values / notes: Matches regular expressions by default; cannot be set together with database.exclude.list. When using name matching mode, values are matched literally.

​
database.exclude.list

Specifies the databases not to capture.
  • Type: LIST
  • Default: No fixed default
  • Importance: High
  • Valid values / notes: Matches regular expressions by default; cannot be set together with database.include.list.

​
collection.include.list

Specifies the collections to capture.
  • Type: LIST
  • Default: No fixed default
  • Importance: High
  • Valid values / notes: Use full database.collection names or patterns; cannot be set together with collection.exclude.list.

​
collection.exclude.list

Specifies the collections not to capture.
  • Type: LIST
  • Default: No fixed default
  • Importance: High
  • Valid values / notes: Use full database.collection names or patterns; cannot be set together with collection.include.list.

​
field.exclude.list

Excludes specified fields from events.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Use comma-separated database.collection.field or nested field names; * is supported for databases and collections.

​
field.renames

Renames fields in events.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Use comma-separated old-field:new-field mappings and full field-qualified names; nested field names must follow the field mapping syntax.

​
Snapshot and Initialization

​
snapshot.mode

Selects how the initial snapshot and subsequent change stream start.
  • Type: STRING
  • Default: initial
  • Importance: Low
  • Valid values / notes: Options are always, initial, never, no_data, initial_only, when_needed, configuration_based, or custom. never is deprecated; use no_data instead. initial_only does not continue capturing changes after the snapshot completes; no_data and never require existing source state and the structural prerequisites needed for decoding.

​
snapshot.collection.filter.overrides

Declares snapshot read filters for specified collections.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Use comma-separated database.collection names; each name must have a corresponding snapshot.collection.filter.overrides.<db>.<collection> property.

​
snapshot.delay.ms

Time to wait before starting the snapshot, in milliseconds.
  • Type: LONG
  • Default: 0
  • Importance: Low
  • Valid values / notes: Must be non-negative; suitable when the source or downstream systems need to finish preparing before initialization.

​
streaming.delay.ms

Time to wait after the snapshot completes before starting to read the change stream, in milliseconds.
  • Type: LONG
  • Default: 0
  • Importance: Low
  • Valid values / notes: Must be non-negative; affects only the wait between the snapshot and continuous capture.

​
snapshot.fetch.size

Batch size used when reading the snapshot.
  • Type: INT
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Must be a non-negative integer; adjust it together with document size, MongoDB load, and Worker memory.

​
snapshot.max.threads

Maximum number of threads used to read collections in parallel during the snapshot.
  • Type: INT
  • Default: 1
  • Importance: Medium
  • Valid values / notes: Must be positive; the actual number of threads does not exceed the number of collections. It increases snapshot parallelism inside one Connector, not the number of Kafka Connect Tasks.

​
snapshot.include.collection.list

Specifies the collections for which a snapshot is required.
  • Type: LIST
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Use a list of regular expressions; it limits snapshot collections and is not equivalent to the capture scope for continuous change streams.

​
incremental.snapshot.chunk.size

Number of documents read in each incremental snapshot chunk.
  • Type: INT
  • Default: 1024
  • Importance: Medium
  • Valid values / notes: Must be non-negative; smaller chunks reduce the pressure from each read but increase incremental snapshot management overhead.

​
incremental.snapshot.watermarking.strategy

Selects the watermarking strategy for incremental snapshots.
  • Type: STRING
  • Default: INSERT_INSERT
  • Importance: Low
  • Valid values / notes: Options are INSERT_INSERT or INSERT_DELETE; signal-driven incremental snapshots require signal support to be enabled.

​
Buffering and Publication

​
max.batch.size

Maximum number of records taken from the internal queue and published in one batch.
  • Type: INT
  • Default: 2048
  • Importance: Medium
  • Valid values / notes: Must be positive and should be less than max.queue.size.

​
max.queue.size

Maximum number of records that the internal event queue can buffer.
  • Type: INT
  • Default: 8192
  • Importance: Medium
  • Valid values / notes: Must be greater than max.batch.size; increasing it can absorb short-term fluctuations but increases memory usage and the amount of data buffered before a failure.

​
max.queue.size.in.bytes

Maximum number of bytes that the internal event queue can buffer.
  • Type: LONG
  • Default: 0
  • Importance: Medium
  • Valid values / notes: Must be non-negative; 0 means that the byte limit is disabled. Use it to constrain memory when MongoDB document sizes vary significantly.

​
poll.interval.ms

Interval at which the Connector polls records from the internal queue, in milliseconds.
  • Type: LONG
  • Default: 500
  • Importance: Medium
  • Valid values / notes: Must be positive; a smaller value usually reduces publication wait time but increases polling frequency.

​
query.fetch.size

Driver fetch size used when reading snapshots or queries.
  • Type: INT
  • Default: 0
  • Importance: Medium
  • Valid values / notes: Must be a non-negative integer; 0 uses the driver or database default behavior.

​
Events and Heartbeats

​
tombstones.on.delete

Whether to send a tombstone record after a delete event.
  • Type: BOOLEAN
  • Default: true
  • Importance: Medium
  • Valid values / notes: true sends DELETE and tombstone records; false sends only DELETE. Select this according to the deletion conventions of downstream compacted topics or state stores.

​
skipped.operations

Specifies operation types not to publish.
  • Type: LIST
  • Default: [t]
  • Importance: Low
  • Valid values / notes: Options are operation codes c, u, d, t, or none; the default skips truncate operations.

​
heartbeat.interval.ms

Interval at which heartbeat events are sent, in milliseconds.
  • Type: INT
  • Default: 0
  • Importance: Medium
  • Valid values / notes: Must be non-negative; 0 disables heartbeat events. Low-volume sources can enable heartbeats to improve progress and offset visibility.

​
heartbeat.topics.prefix

Name prefix for heartbeat topics.
  • Type: STRING
  • Default: __debezium-heartbeat
  • Importance: Low
  • Valid values / notes: Requires particular attention only when heartbeat events are enabled; it should match topic naming and permission policies.

​
extended.headers.enabled

Whether to write Debezium extended context headers to source records.
  • Type: BOOLEAN
  • Default: true
  • Importance: Low
  • Valid values / notes: When disabled, downstream systems cannot rely on these extended headers to obtain the corresponding context.

​
Signals and Incremental Snapshots

​
signal.data.collection

MongoDB database and collection that store Connector signals.
  • Type: STRING
  • Default: No fixed default
  • Importance: Medium
  • Valid values / notes: Format is database.collection; signal processing is disabled when it is not set. Create this collection and grant the required permissions before using incremental snapshots.

​
signal.poll.interval.ms

Interval at which the signal collection is polled, in milliseconds.
  • Type: LONG
  • Default: 5000
  • Importance: Medium
  • Valid values / notes: Must be positive; a smaller value detects signals faster but increases source read frequency.

​
signal.enabled.channels

List of enabled signal channels.
  • Type: LIST
  • Default: [source]
  • Importance: Medium
  • Valid values / notes: The source channel is enabled by default; add other channels only when the corresponding signal source and permissions are configured.

​
Topics, Schema, and Extensions

​
topic.naming.strategy

Strategy class used to generate Kafka topic names.
  • Type: CLASS
  • Default: io.debezium.schema.DefaultTopicNamingStrategy
  • Importance: Medium
  • Valid values / notes: The class must implement TopicNamingStrategy; changing the strategy changes topic routing and may affect downstream subscriptions and historical data compatibility.

​
schema.name.adjustment.mode

Mode for adjusting schema names to support specific serialization formats.
  • Type: STRING
  • Default: none
  • Importance: Low
  • Valid values / notes: Options are none, avro, or avro_unicode; confirm the naming requirements of the downstream Schema Registry or Converter before selecting one.

​
sourceinfo.struct.maker

Implementation class that creates the MongoDB Source information structure.
  • Type: CLASS
  • Default: io.debezium.connector.mongodb.MongoDbSourceInfoStructMaker
  • Importance: Low
  • Valid values / notes: A custom class must provide a compatible SourceInfoStructMaker implementation; retain the default unless there is a clear need to extend event metadata.

​
converters

Registers custom Converters and their configuration prefixes.
  • Type: STRING
  • Default: No fixed default
  • Importance: Low
  • Valid values / notes: Use comma-separated custom Converter names and provide the corresponding type and options for each; custom classes must be visible in the Worker plugin path.

​
post.processors

Registers event post-processors and their configuration prefixes.
  • Type: STRING
  • Default: No fixed default
  • Importance: Low
  • Valid values / notes: Use comma-separated post-processor names and provide each type and its options; post-processors change the content or metadata of published events.

​
Error Handling

​
event.processing.failure.handling.mode

How to handle corrupted or unparseable events.
  • Type: STRING
  • Default: fail
  • Importance: Medium
  • Valid values / notes: Options are fail, warn, or ignore. warn and ignore may skip events that cannot be processed and should be used together with error logging and business data-integrity checks.

​
errors.max.retries

Maximum number of retries for retryable source-side errors.
  • Type: INT
  • Default: -1
  • Importance: Low
  • Valid values / notes: -1 means unlimited retries, 0 disables retries, and a positive number specifies a finite limit; retries do not provide an end-to-end no-duplicate delivery guarantee.

​
errors.retry.timeout

Total time for the Kafka Connect Worker to retry failed records or tasks, in milliseconds.
  • Type: LONG
  • Default: 0
  • Importance: Medium
  • Valid values / notes: 0 disables this retry window and -1 means unlimited retries; this is part of the Worker error policy and should be planned together with task-failure alerts.

​
errors.tolerance

Kafka Connect Worker’s error tolerance level.
  • Type: STRING
  • Default: none
  • Importance: Medium
  • Valid values / notes: Options are none or all; all skips problematic records and cannot replace error analysis and compensation procedures.

​
Kafka Connect Framework

​
connector.class

Specifies the Connector class that Kafka Connect instantiates.
  • Type: STRING
  • Default: No fixed default
  • Importance: High
  • Required: Yes
  • Valid values / notes: Use io.debezium.connector.mongodb.MongoDbConnector.

​
tasks.max

Maximum number of Kafka Connect Tasks that the Connector can use.
  • Type: INT
  • Default: 1
  • Importance: High
  • Valid values / notes: Must be at least 1; this Connector returns one Task configuration for MongoDB replication, and increasing this value does not create an independent Kafka Connect Task for each replica set. Replica set replication and snapshot parallelism are handled by internal Connector threads.

​
key.converter

Converter class used to serialize source record Keys.
  • Type: CLASS
  • Default: No fixed default
  • Importance: Low
  • Valid values / notes: Must implement the Kafka Connect Converter; when not set at the Connector level, the Worker default is used.

​
value.converter

Converter class used to serialize MongoDB CDC event Values.
  • Type: CLASS
  • Default: No fixed default
  • Importance: High
  • Valid values / notes: Must implement the Kafka Connect Converter; it should match the deserialization method used by downstream consumers.

​
Best Practices

​
Limit the Initialization Scope and Continue Incremental Synchronization on First Connection

Applicable business scenario: When connecting existing MongoDB data for the first time, establish a baseline for specified databases and collections before continuing to receive subsequent changes, avoiding publication of existing data from unrelated collections. Configuration example:
Key notes: Confirm the collection scope, source load during the snapshot, and downstream capacity before starting the Connector. When no usable prior offset exists, the default snapshot.mode=initial publishes READ records for matching collections and continues capturing changes; newly added collections do not automatically have their existing data published, so use a separate snapshot plan when that is required. collection.include.list and collection.exclude.list cannot be configured together.

​
Ingest Only Subsequent Changes When a Data Baseline Already Exists

Applicable business scenario: When the downstream system already has a data baseline created through backup, batch processing, or another method, receive only subsequent MongoDB changes from the Connector’s starting position instead of publishing existing documents again. Configuration example:
Key notes: no_data does not send snapshot records for existing documents, but the downstream system must still have the structure and baseline required to parse subsequent events. Before switching to this mode, identify the point in time represented by the downstream baseline and preserve the stable topic.prefix and Worker offsets; do not use it to bypass an incomplete initial snapshot or recover expired MongoDB change history.

​
Improve Progress Visibility for Continuously Running Low-Volume Sources

Applicable business scenario: When business collections receive few writes during normal operation but the Connector must still be shown to be reading continuously and maintaining source progress, distinguish “no changes” from “not running” in downstream monitoring. Configuration example:
Key notes: Heartbeat events improve progress and offset visibility during low-volume periods; they do not indicate that a business change occurred at the source. Monitor the Task status, heartbeat topic, and Worker offsets together; after a restart, recovery still depends on persisted offsets and MongoDB change stream history, and heartbeats do not provide an end-to-end exactly-once guarantee.

​
Monitoring

​
What to Monitor

Monitor the health status and restart counts of the Kafka Connect Worker, Connector, and Tasks; check source-to-Kafka throughput, end-to-end latency, backlog, offset commit times, and commit failures; also monitor MongoDB connection errors, retries, task error logs, event processing failures, Kafka publication failures, and Worker JVM heap, GC, thread, and CPU signals. When errors.tolerance=all or DLQ-related error handling is enabled, also monitor skipped records and DLQ traffic to avoid mistaking error tolerance for normal synchronization.

​
Import the Grafana Dashboard

Download the AutoMQ Connect Grafana Dashboard, select in Grafana the data source that collects Kafka Connect metrics, confirm that the cluster, Worker, Connector, and Task labels match the metrics collection configuration, and upload the JSON through the Dashboards import function while mapping the data source.

​
Limitations

  • Standalone MongoDB does not provide the Change Streams required by this Connector and therefore cannot be used as this Connector’s change stream source.
  • Increasing tasks.max does not split replication by replica set into multiple Kafka Connect Tasks; replica set replication uses internal Connector threads, and snapshot threads do not change the number of Tasks.
  • initial_only performs only the snapshot and does not continue capturing changes after the snapshot ends.
  • no_data and the deprecated never do not publish snapshot records for existing documents and require existing source state and the structural prerequisites needed by downstream decoding.
  • Incremental snapshot window deduplication handles collisions between the snapshot and concurrent changes; it is not an end-to-end exactly-once guarantee and does not represent a single global point-in-time snapshot.
  • The change stream pipeline cannot use MongoDB indexes; regular-expression scope filtering also adds internal field-transformation overhead.
  • During connection restart or offset recovery, if MongoDB has cleaned up the retained change stream history, the Connector cannot replay all intermediate events without gaps from the old position.
  • The internal buffer is not a persistent checkpoint; increasing the queue changes only memory, backpressure, and latency behavior and does not change delivery guarantees.
  • If a failure occurs after Kafka publication completes but before the Worker offset is persisted, a restart may publish events that were already sent; downstream systems should handle duplicate records.

​
FAQ

​
Why does the Connector fail to connect to MongoDB?

Check whether MongoDB is a replica set or cluster deployment that supports Change Streams, whether the connection string contains the correct host, port, and replica set information, whether the authentication database is correctly specified by mongodb.authsource, and whether the account can read the target databases. When TLS is enabled, also check the truststore, certificate hostname, and Worker file permissions; do not bypass certificate hostname validation in production by setting mongodb.ssl.invalid.hostname.allowed=true.

​
Why are no existing documents received after startup?

Confirm that snapshot.mode is not set to no_data, never, or another mode that does not perform a data snapshot, and check whether database.include.list, collection.include.list, the exclude lists, or snapshot.include.collection.list exclude the target collections. If the downstream system has already stored offsets for the same topic.prefix, restarting does not automatically resend the full existing dataset; re-plan based on the data baseline and snapshot strategy.

​
Why are events received from only some collections?

Check the database and collection include lists, exclude lists, and their regular expressions, and confirm that include and exclude are not both configured at the same level. Also check whether snapshot.include.collection.list limits only the snapshot, so that the snapshot scope is not confused with the continuous change stream scope.

​
Why does an empty-value record appear after a delete event?

This is the default behavior of tombstones.on.delete=true: the Connector publishes DELETE first and then a tombstone with the same Key, allowing Kafka log compaction to remove the old value. If downstream systems do not accept tombstones, set tombstones.on.delete=false, but evaluate the downstream deletion and state-cleanup logic at the same time.

​
Why do duplicate events appear after a restart?

Kafka Connect offset persistence may occur after event publication; if a failure occurs between publication and offset flush, recovery may republish events from an earlier position. Confirm that topic.prefix, Worker offset storage, and MongoDB change history remain unchanged, and have downstream systems handle duplicates using stable document Keys, source positions, or business idempotency logic. The Connector does not promise end-to-end exactly-once delivery.

​
Why did no incremental snapshot run after it was enabled?

Confirm that signal.data.collection is set, that the collection exists and the account can read the signal collection, that signal.enabled.channels includes source, and that the incremental snapshot configuration uses a supported watermarking strategy. Also confirm that the signal document uses the correct database and collection names and that the related collections have document keys that the Connector can read.