mongodb_cdc
Streams changes from a MongoDB replica set.
Introduced in version 4.48.0.
Read from a MongoDB replica set using change streams. It’s only possible to watch for changes when using a sharded MongoDB or a MongoDB cluster running as a replica set. Use collections to choose which collections in the database to stream changes from.
If a stored resume position can no longer be resumed (for example it has aged out of the oplog), the input clears its checkpoint and re-runs the snapshot rather than retrying a dead position. After 2 such recoveries in a row without the change stream ever advancing in between - which usually means the snapshot is taking longer than the oplog window - the next failure keeps the checkpoint and the input fails on every reconnect with an actionable error instead of re-snapshotting forever; after addressing the cause, restart the pipeline (or delete the checkpoint cache entry) to resume recovery.
With neither stream_snapshot nor incremental_snapshot.enabled set, there is no snapshot to re-run, so recovery would have to skip the changes between the lost position and now. The input refuses to do that silently: by default it fails with an error on every reconnect and preserves the checkpoint, and skipping the gap has to be opted into with on_unresumable_position: reset.
By default MongoDB does not propagate changes in all cases. In order to capture all changes (including deletes) in a MongoDB cluster one needs to enable pre and post image saving and the collections also need to enable saving these pre and post images. For more information see MongoDB documentation and the document_mode options.
Snapshot and streaming modes
When stream_snapshot is true and checkpoint_cache holds no stream position, this input records the current change stream position, reads all documents in the selected collections using snapshot_parallelism connections, and then streams the changes made since the recorded position. Snapshot progress is not checkpointed, so if the pipeline restarts before the snapshot completes, the snapshot runs again from the start. Once the snapshot completes and is acknowledged, its position is stored in checkpoint_cache, so later restarts resume the stream without another snapshot.
When snapshot_parallelism is greater than 1, this input reads the snapshot in parallel by splitting each collection into chunks with the splitVector command. In environments where splitVector is not available, such as MongoDB Atlas, set snapshot_auto_bucket_sharding to true to compute the chunks with the $bucketAuto aggregation stage instead.
When stream_snapshot is false and no position is stored, this input starts streaming from the latest position in the operations log (oplog). In both modes, a restart resumes streaming from the position stored in checkpoint_cache.
Incremental snapshots
With incremental_snapshot.enabled: true the snapshot no longer holds back streaming. This is mutually exclusive with stream_snapshot: both read the same documents, so enabling both would deliver everything twice. On a first start the input streams changes immediately and reads every collection in incremental_snapshot.chunk_size chunks, in _id order, between them. Each chunk is released once the change stream has caught up with the moment it was read, and any document the stream changed meanwhile is dropped from the chunk, so a document is never emitted older than a change already streamed for it. Chunks are read from the primary whatever the read preference in url, which still applies to the change stream: a lagging secondary could return documents older than changes already streamed. Progress is stored in checkpoint_cache under incremental_snapshot.checkpoint_cache_key, so a restart, for example after a credential expiry, continues from the last acknowledged chunk rather than starting over. The progress is written again with every stream position, so a cache with a TTL expires both together.
A collection added to collections is read on the next start, while the others carry on streaming. Turning on incremental_snapshot.enabled in a pipeline that already has a stream position stored, whether it ran with stream_snapshot or without it, reads nothing: with a stored position and no stored snapshot progress, every collection counts as read. To backfill such a pipeline, set signal_collection_name and send a snapshot-execute signal. A snapshot-execute signal reads a collection on request, in every snapshot mode: the same chunked read also serves pipelines running with stream_snapshot or with no snapshot at all.
Documents from a chunk have the operation set to "read" and the operation time of the chunk read. Collections are read one at a time, so snapshot_parallelism and snapshot_auto_bucket_sharding do not apply. Chunks are bounded by index ranges rather than comparisons, so a collection whose _id values mix BSON types is read in full.
A document can be delivered twice, once streamed and once in a chunk: when it is inserted after streaming starts and before its chunk is read. Treat documents as idempotent upserts keyed by _id, as is standard CDC practice. With document_mode: partial_update, an update streamed while its document is buffered replaces the chunk’s full document with the update description alone, so a consumer that has not seen the document before receives only the delta; prefer update_lookup or pre_and_post_images with incremental snapshots. On sharded collections documents are matched on _id alone. MongoDB does not enforce a unique _id across shards unless _id is the shard key, so make sure your application keeps it unique. Otherwise the snapshot can skip documents that share an _id.
A chunk can only be released once the change stream has passed the moment it was read. On a busy deployment that happens almost at once, and on a quiet replica set the input releases several chunks per poll. A quiet sharded cluster is slower: its change stream only advances as far as its least active shard, which moves on with its periodic no-op write, every 10 seconds by default (the shards' periodicNoopIntervalSecs). Expect about one chunk per interval there, and raise incremental_snapshot.chunk_size to compensate, with checkpoint_limit raised to at least twice that.
Progress is reported through metrics: mongodb_cdc_incremental_snapshot_collections_remaining counts the collections still to read, mongodb_cdc_incremental_snapshot_documents_released and mongodb_cdc_incremental_snapshot_chunks_released count what has been released per collection, and mongodb_cdc_incremental_snapshot_chunks_deferred counts chunk reads deferred after a transient failure. The start and end of each collection’s read are logged at info level.
Signals
When signal_collection_name is set, documents inserted into that collection control the input while it runs. Each signal has a type and a data field, which may be an embedded document or a JSON string:
-
log: logsdata.messageat info level, for example{"type": "log", "data": {"message": "hello"}}. -
snapshot-execute: reads the named collections incrementally, alongside streaming, for example{"type": "snapshot-execute", "data": {"collections": ["orders"]}}(tablesis accepted as an alias ofcollections). Each named collection must also be incollections, because the input only watches those for changes. If a signal names any other collection, the input skips the whole signal and logs an error. A collection already read is read again, which makes the signal a way to re-read a collection on demand; one still queued or being read is not queued twice.The read works the same in every snapshot mode. It does not need
incremental_snapshot.enabled: withstream_snapshot, or with no snapshot at all, the named collections are read inincremental_snapshot.chunk_sizechunks between change events, exactly as described under incremental snapshots, and the sameretry_cooldownand duplicate-delivery notes apply. Nothing is read until a signal asks for it, a collection added tocollectionslater included: onlyincremental_snapshot.enabledreads one on the next start. Progress is stored underincremental_snapshot.checkpoint_cache_keyalongside the stream position, so a restart continues a requested read from the last acknowledged chunk rather than losing the request.
Malformed signals and unknown types are logged and skipped. Updates and deletes in the signal collection are ignored.
Signal inserts are emitted as regular messages (operation=insert, collection=<signal_collection_name>), the same as postgres_cdc and aws_dynamodb_cdc emit their signals. To keep them out of the output, filter on the collection metadata field:
pipeline:
processors:
- mapping: |
root = if @collection == "rpcn_signals" { deleted() } else { this }Scaling
All configured collections are consumed through a single database-level change stream: the collection list is a server-side filter on that one stream, not a set of independent streams. Because a change stream is one totally-ordered cursor over the replica set’s oplog, throughput does not increase with additional collections, and CPU beyond roughly two cores is not used. This is a property of MongoDB change streams rather than of this input. To scale beyond the single-stream ceiling, shard the cluster: a change stream against a sharded cluster merges parallel per-shard cursors on the server side.
Size the oplog for consumer lag: a change stream resumes from a position in the oplog, so if this input falls behind, or is stopped, for longer than the oplog retains data, the resume position is lost and the stream is invalidated. Ensure the oplog window comfortably exceeds the longest expected lag or downtime under peak write load.
Metadata
Each message emitted by this plugin has the following metadata:
-
operation: Eitherinsert,replace,delete, orupdatefor changes streamed. Documents from a snapshot have the operation set toread. -
collection: The collection the document was written to. -
operation_time: The oplog time for when this operation occurred. -
schema: The collection schema in benthos common schema format (set as immutable metadata). Extracted from the collection’s$jsonSchemavalidator if available, otherwise inferred from the first document seen. Not present on messages where no schema could be determined (for example, deletes without pre-images when no prior schema is cached).
Schema Detection
Schema metadata is discovered using a two-tier strategy:
-
$jsonSchema validators are preferred and queried at startup for each watched collection. When a validator exists, the schema provides accurate type information and required/optional field classification.
-
When no validator exists, schema is inferred from the first document received per collection. All fields are marked optional.
Change detection: when a document’s top-level field set differs from the cached schema, the schema is re-inferred from that document. This applies to both validator-sourced and inference-sourced schemas.
Limitations: type changes within existing fields and structural changes inside nested subdocuments are not detected automatically. Restart the input to force a full schema refresh.
Fields with null values, unknown BSON types, or mixed-type arrays are mapped to the Any schema type. The parquet_encode processor does not support Any and will error if it encounters one. Add an upstream processor (for example, mapping) to convert or remove these fields before parquet_encode.
Schema stability: MongoDB collections may contain documents with varying field sets. When this occurs, the schema updates on each structural change, which can cause frequent schema version bumps in schema registries with compatibility modes. For schema registry targets, configuring a $jsonSchema validator on the collection is strongly recommended.
-
Common
-
Advanced
input:
label: ""
mongodb_cdc:
url: "" # No default (required)
database: "" # No default (required)
username: ""
password: ""
collections: [] # No default (required)
checkpoint_key: mongodb_cdc_checkpoint
checkpoint_cache: "" # No default (required)
checkpoint_interval: 5s
checkpoint_limit: 1000
read_batch_size: 1000
read_max_wait: 1s
stream_snapshot: false
snapshot_parallelism: 1
signal_collection_name: "" # No default (optional)
auto_replay_nacks: trueinput:
label: ""
mongodb_cdc:
url: "" # No default (required)
database: "" # No default (required)
username: ""
password: ""
aws:
enabled: false
region: "" # No default (optional)
session_duration: 1h
id: "" # No default (optional)
secret: "" # No default (optional)
token: "" # No default (optional)
role: "" # No default (optional)
role_external_id: "" # No default (optional)
roles: [] # No default (optional)
collections: [] # No default (required)
checkpoint_key: mongodb_cdc_checkpoint
checkpoint_cache: "" # No default (required)
checkpoint_interval: 5s
checkpoint_limit: 1000
checkpoint_write_timeout: 10s
on_unresumable_position: fail
read_batch_size: 1000
read_max_wait: 1s
stream_snapshot: false
snapshot_parallelism: 1
snapshot_auto_bucket_sharding: false
incremental_snapshot:
enabled: false
chunk_size: 500
checkpoint_cache_key: ""
retry_cooldown: 5s
signal_collection_name: "" # No default (optional)
document_mode: update_lookup
json_marshal_mode: canonical
app_name: benthos
auto_replay_nacks: truePrerequisites
-
MongoDB version 6 or later
-
Network access from the cluster where your Redpanda Connect pipeline is running to the source database environment.
-
A MongoDB database running as a replica set or in a sharded cluster using replica set protocol version 1.
-
A MongoDB database using the WiredTiger storage engine.
Enable connectivity from cloud-based data sources
To establish a secure connection between a cloud-based data source and Redpanda Connect, you must add the IP addresses of your Redpanda Connect instances to your firewall rules.
Fields
auto_replay_nacks
Whether to automatically replay rejected messages (negative acknowledgements, or nacks) at the output level. If the cause of rejections persists, leaving this option enabled can result in back pressure.
Set auto_replay_nacks to false to delete rejected messages. Disabling auto replays can greatly improve memory efficiency of high throughput streams, as the original shape of the data is discarded immediately upon consumption and mutation.
Type: bool
Default: true
aws
AWS IAM authentication using the MONGODB-AWS mechanism, for example against MongoDB Atlas. When enabled, IAM credentials are used instead of a static username and password. Role-derived session credentials are resolved when the component connects and are re-resolved whenever it reconnects. The mongodb processor and cache establish their client once at creation and cannot refresh expiring session credentials, so role, roles and session tokens are rejected for those components; use the ambient credential chain or long-lived access keys with them. For long-running pipelines, prefer the ambient credential chain (leave keys and roles unset), which the driver refreshes automatically.
Requires version 4.106.0 or later.
Type: object
aws.enabled
Enable AWS IAM authentication using the driver-native MONGODB-AWS mechanism. The MongoDB Atlas database user must be created with the AWS IAM authentication type, and connections require TLS. When no static credentials or roles are configured, the ambient AWS credential chain (environment variables, EC2 instance profile, EKS pod role) is used and expiring credentials are refreshed automatically.
Type: bool
Default: false
aws.id
The AWS access key ID to authenticate with. When empty, the default AWS credential chain is used.
Type: string
aws.region
The AWS region used when assuming roles (for STS calls). Only used when role or roles are configured; the ambient and static-key paths ignore it. If no region is specified then the environment default is used.
Type: string
aws.role
Optional AWS IAM role ARN to assume for authentication. Cannot be combined with roles; use the roles array instead when chaining multiple roles.
Type: string
aws.role_external_id
Optional external ID to use when assuming the role set in role. Each entry in roles sets its own external ID.
Type: string
aws.roles[]
Optional array of AWS IAM roles to assume for authentication. Roles are assumed in sequence, each using the credentials of the previous one, enabling chaining for purposes such as cross-account access. Each role can optionally specify an external ID. Cannot be combined with role.
Type: array<object>
aws.secret
The AWS secret access key that pairs with id.
|
This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets. |
Type: string
aws.session_duration
The duration of the STS session requested when assuming roles. AWS requires at least 15 minutes and caps sessions created through role chaining at one hour. Only used when role or roles are configured. For long-running pipelines, prefer the ambient credential chain over a fixed session duration, since the driver refreshes ambient credentials automatically as they near expiry. When using role assumption with this input, credentials are freshly resolved after the initial snapshot completes, so the streaming phase starts with a full session. With stream_snapshot, the snapshot itself must still complete within a single session duration: snapshot progress is not checkpointed, so a credential expiry mid-snapshot restarts the snapshot from scratch after reconnecting. With incremental_snapshot.enabled, snapshot progress is checkpointed, so a credential expiry does not restart the snapshot: after reconnecting, the input continues from the last acknowledged chunk. Once the snapshot completes and is fully acknowledged, its position is checkpointed, so later restarts resume the stream without re-running the snapshot.
Type: string
Default: 1h
aws.token
The AWS session token to use with id and secret. Required only when using short-term credentials.
Type: string
checkpoint_cache
Specify a cache resource to store the oplog position for the most recent data update streamed to Redpanda Connect. After a restart, Redpanda Connect can continue processing changes from this position, avoiding the need to reprocess all collection updates.
Type: string
checkpoint_key
The key identifier used to store the stream’s resume position in checkpoint_cache. If you have multiple mongodb_cdc inputs sharing the same cache, you can provide an alternative key.
Type: string
Default: mongodb_cdc_checkpoint
checkpoint_limit
The maximum number of in-flight messages emitted from this input. Increasing this limit enables parallel processing, and batching at the output level. To preserve at-least-once guarantees, the stream’s resume position is only checkpointed once all messages before it are delivered.
Type: int
Default: 1000
checkpoint_write_timeout
Bounds the checkpoint writes that run outside the normal read loop - storing the position a completed snapshot reached, and clearing a position that can no longer be resumed from - so that a shutdown racing either of them is not extended indefinitely by a slow cache. Raise this for slow remote caches (for example redis or dynamodb), where losing the post-snapshot write costs a full re-snapshot on the next start.
Requires version 4.106.0 or later.
Type: string
Default: 10s
collections[]
A list of collections to stream changes from. Specify each collection name as a separate item.
Type: array<string>
document_mode
The mode in which MongoDB emits document changes to Redpanda Connect, specifically updates and deletes.
Type: string
Default: update_lookup
| Option | Summary |
|---|---|
|
In this mode update operations only have a description of the update operation, which follows the following schema: { "_id": <document_id>, "operations": [ # type == set means that the value was updated like so: # root.foo."bar.baz" = "world" {"path": ["foo", "bar.baz"], "type": "set", "value":"world"}, # type == unset means that the value was deleted like so: # root.qux = deleted() {"path": ["qux"], "type": "unset", "value": null}, # type == truncatedArray means that the array at that path was truncated to value number of elements # root.array = this.array.slice(2) {"path": ["array"], "type": "truncatedArray", "value": 2} ] } |
|
Uses pre and post image collection to emit the full documents for update and delete operations. To use and configure this mode see the setup steps in the MongoDB documentation. |
|
In this mode insert, replace and update operations have the full document emitted and deletes only have the _id field populated. Documents updates lookup the full document. This corresponds to the updateLookup option, see the MongoDB documentation for more information. |
incremental_snapshot
Configures chunked snapshotting that runs alongside change streaming. With enabled every collection is read this way on the first start; the other settings also govern the reads a snapshot-execute signal requests through signal_collection_name, which work in every snapshot mode.
Requires version 4.116.0 or later.
Type: object
incremental_snapshot.checkpoint_cache_key
The key used to store the incremental snapshot’s progress in checkpoint_cache, alongside the stream position. Defaults to checkpoint_key with _incremental_snapshot appended, so inputs with distinct checkpoint_key values never share progress.
If you delete the stored progress, or change this key, while a stream position is stored, a restart treats every collection as read. To read a collection again, send a snapshot-execute signal. Unlike postgres_cdc, a new key does not cause a new read. The reads a signal requests store their progress here too, whether or not enabled is set. While enabled is false and no signal_collection_name is set, nothing keeps the progress current, so the input discards what is left under this key and enabling either later never resumes a read the stream has long since moved past.
Type: string
Default: ""
incremental_snapshot.chunk_size
The number of documents read per chunk. A released chunk is in flight as a whole, so it must not exceed checkpoint_limit, and keeping it well below leaves room for streamed changes and the next chunk alongside it.
Type: int
Default: 500
incremental_snapshot.enabled
Streams changes from the start and reads the collections in chunk_size chunks, in _id order, alongside them. Each chunk is released once the change stream has caught up with the moment it was read, and progress is stored in checkpoint_cache, so a restart continues from the last acknowledged chunk. Unlike stream_snapshot this needs no up-front snapshot phase and does not delay streaming. The two are mutually exclusive: both read the same documents, so enabling both would deliver everything twice.
Type: bool
Default: false
incremental_snapshot.retry_cooldown
How long the snapshot waits before retrying a chunk read that failed transiently, for example during a replica set election or after a dropped connection. Change events keep streaming meanwhile. Setting it to 0s retries on the next streamed change.
Type: string
Default: 5s
json_marshal_mode
Controls the format used to convert a message from BSON to JSON when it is received by Redpanda Connect.
Type: string
Default: canonical
| Option | Summary |
|---|---|
|
A string format that emphasizes type preservation at the expense of readability and interoperability. That is, conversion from canonical to BSON will generally preserve type information except in certain specific cases. |
|
A string format that emphasizes readability and interoperability at the expense of type preservation.That is, conversion from relaxed format to BSON can lose type information. |
on_unresumable_position
What to do when the stored stream position can no longer be resumed from (for example it has aged out of the oplog) and no snapshot is enabled, so there is no snapshot to recover with: fail stops the input with an error, preserving the checkpoint for inspection; reset clears the checkpoint and restarts streaming from the current oplog position, skipping the changes between the lost position and now. When stream_snapshot or incremental_snapshot.enabled is enabled this field has no effect: recovery re-runs the snapshot. This recovers the current state of every document, but not the deletes between the lost position and now.
Requires version 4.106.0 or later.
Type: string
Default: fail
Options: fail, reset
password
The password to use for authentication. Used together with username for basic authentication.
|
This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Secrets. |
Type: string
Default: ""
read_batch_size
The number of documents to fetch in each message batch from MongoDB.
Type: int
Default: 1000
read_max_wait
The maximum duration MongoDB waits to accumulate read_batch_size documents on a change stream before returning the batch to Redpanda Connect.
Type: string
Default: 1s
signal_collection_name
A collection in database to watch for signals, documents that control the input while it runs. Each signal has a type and a data field: log logs data.message, and snapshot-execute reads the collections named in data.collections in incremental_snapshot.chunk_size chunks alongside streaming, for example {"type": "snapshot-execute", "data": {"collections": ["orders"]}}, in every snapshot mode: incremental_snapshot.enabled is not required. Signal inserts are also emitted as regular messages with the collection metadata set to this collection, so filter them out downstream if they are not wanted.
Requires version 4.116.0 or later.
Type: string
snapshot_auto_bucket_sharding
If set to true, uses the $bucketAuto aggregation stage instead of the default splitVector command to split the snapshot data into chunks for processing. This allows parallel collection reading in environments where privileged access to the MongoDB cluster is not allowed, such as MongoDB Atlas, where the splitVector command is not available. To enable parallel processing in these environments:
-
Set this field to
true. -
Set
stream_snapshottotrue. -
Increase
snapshot_parallelismto a value greater than1.
Type: bool
Default: false
snapshot_parallelism
Specifies the number of connections to use when reading the initial snapshot from one or more collections. Increase this number to enable parallel processing of the snapshot.
This feature uses the splitVector command to split snapshot data into chunks for more efficient processing.
This field is only applicable when stream_snapshot is set to true.
Type: int
Default: 1
