Cloud
Also in Self-Managed

mongodb_cdc

Streams data changes from a MongoDB replica set, using MongoDB’s change streams to capture data updates.

  • Common

  • Advanced

inputs:
  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
    auto_replay_nacks: true
inputs:
  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
    document_mode: update_lookup
    json_marshal_mode: canonical
    app_name: benthos
    auto_replay_nacks: true

Prerequisites

  • MongoDB version 6 or later

  • Network access from the cluster where your Redpanda Connect pipeline is running to the source database environment. For detailed networking information, including how to set up a VPC peering connection, see Redpanda Cloud Networking.

  • 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 (BYOC)

To establish a secure connection between a cloud-based data source and Redpanda Connect, you must add the NAT Gateway IP address of your Redpanda cluster to the allowlist of your data source.

Data capture method

The mongodb_cdc input uses change streams to capture data changes, which does not propagate all changes to Redpanda Connect. To capture all changes in a MongoDB cluster, including deletions, enable pre- and post-image saving for the cluster and required collections. For more information, see document_mode options and the MongoDB documentation.

Data replication

Redpanda Connect allows you to specify which database collections in your source database to receive changes from.

You can also run the mongodb_cdc input in one of two modes, depending on whether you need a snapshot of existing data before streaming updates.

  • Snapshot mode: Redpanda Connect first captures a snapshot of all data in the selected collections and streams the contents before processing changes from the last recorded operations log (oplog) position.

  • Streaming mode: Redpanda Connect skips the snapshot and processes only the most recent data changes, starting from the latest oplog position.

Snapshot mode

If you set the stream_snapshot field to true, Redpanda Connect connects to your MongoDB database and does the following to capture a snapshot of all data in the selected collections:

  1. Records the latest oplog position.

  2. Determines the strategy for splitting the snapshot data down into shards or chunks for more efficient processing:

    1. If snapshot_auto_bucket_sharding is set to false, the internal $splitVector command is used to compute shards.

    2. If snapshot_auto_bucket_sharding is set to true, the $bucketAuto command is used instead. This setting is for environments, such as MongoDB Atlas, where the $splitVector command is not available.

  3. This input then uses the number of connections specified in snapshot-parallelism to read the selected collections.

    If the pipeline restarts during this process, Redpanda Connect must start the snapshot capture from scratch to store the current oplog position in the checkpoint_cache.

  4. Finally, the input uses the stored oplog position to catch up with changes that occurred during snapshot processing.

Streaming mode

If you set the stream_snapshot field to false, Redpanda Connect connects to your MongoDB database and starts processing data changes from the latest oplog position. If the pipeline restarts, Redpanda Connect resumes processing updates from the last oplog position written to the checkpoint_cache.

Metadata

Each message emitted by this plugin has the following metadata:

  • operation: either "insert", "replace", "delete" or "update" for changes streamed. Documents from the initial snapshot have the operation set to "read".

  • 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 $jsonSchema validator if available, otherwise inferred from the first document seen. Not present on messages where no schema could be determined (e.g. deletes without pre-images when no prior schema is cached).

Fields

app_name

The client application name.

Type: string

Default: benthos

auto_replay_nacks

Whether to automatically replay rejected messages (negative acknowledgements) at the output level. If the cause of rejections is persistent, 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.

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 ID of credentials to use.

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 for the role assumption. Only used with the role field, which cannot be combined with roles.

Type: string

aws.roles[]

Optional array of AWS IAM roles to assume for authentication. Roles can be assumed in sequence, 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.roles[].role

AWS IAM role ARN to assume.

Type: string

Default: ""

aws.roles[].role_external_id

Optional external ID for the role assumption.

Type: string

Default: ""

aws.secret

The secret for the credentials being used.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

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. When using mongodb_cdc with role assumption, credentials are freshly resolved after the initial snapshot completes, so the streaming phase starts with a full session. 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. Once the snapshot completes and is fully acknowledged, its position is checkpointed, so later restarts resume the stream without re-running the snapshot. For very large snapshots prefer the ambient credential chain.

Type: string

Default: 1h

aws.token

The token for the credentials being used, required 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_interval

The interval between writing checkpoints to the cache.

Type: string

Default: 5s

checkpoint_key

The key identifier used to store the oplog 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, any given oplog position is not acknowledged until all messages under that offset 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.

Type: string

Default: 10s

collections[]

A list of collections to stream changes from. Specify each collection name as a separate item.

Type: array<string>

database

The name of the MongoDB database to stream changes from.

Type: 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

partial_update

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} ] }

pre_and_post_images

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.

update_lookup

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.

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

canonical

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.

relaxed

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 stream_snapshot is disabled, 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 is enabled this field has no effect: recovery re-runs the snapshot, which loses nothing.

Type: string

Default: fail

Options: fail, reset

password

The password to connect to the database.

This field contains sensitive information that usually shouldn’t be added to a configuration directly. For more information, see Manage Secrets before adding it to your configuration.

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 the read_batch_size documents on a change stream before returning the batch to Redpanda Connect.

Type: string

Default: 1s

snapshot_auto_bucket_sharding

Uses the $bucketAuto command instead of the default, $splitVector, to split the snapshot data into chunks for processing. This is required for environments, such as MongoDB Atlas, where the $splitVector command is not available. To enable parallel processing in these environments:

  • Set this field to to true.

  • Set stream_snapshot to true.

  • Increase snapshot_parallelism to a value greater than 1.

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

stream_snapshot

When set to true, this input streams a snapshot of all existing data in the source collections before streaming data changes.

Type: bool

Default: false

url

The URL of the target MongoDB server.

Type: string

# Examples:
url: mongodb://localhost:27017

username

The username to connect to the database.

Type: string

Default: ""