Connect Cdc Mongodb

Redpanda Connect CDC: MongoDB

The mongodb_cdc input in Redpanda Connect streams change data capture (CDC) from a MongoDB database into Redpanda or any Kafka-compatible topic. It uses MongoDB Change Streams (requires MongoDB 4.0+ running as a replica set or sharded cluster), and optionally snapshots all existing documents before switching to live change streaming. This is an Enterprise feature — a Redpanda Enterprise license is required.

The connector tracks its position in the change stream using a resume token, which is stored in a configured cache resource. On restart, it resumes from the last stored token rather than replaying from the beginning.

Quickstart

1. Verify your MongoDB topology

Change Streams require a replica set or sharded cluster — they are not available on standalone mongod instances.

# Connect via mongosh and check replication status
mongosh "mongodb://localhost:27017"
rs.status()   # must show a replica set configuration

# Create a user with read on the target DB
# (covers find + listCollections + changeStream action)
use admin
db.createUser({
  user: "cdc_user",
  pwd:  "secret",
  roles: [
    { role: "read", db: "mydb" }
  ]
})

2. Full pipeline YAML (snapshot + stream, two collections)

# mongodb-cdc-pipeline.yaml
cache_resources:
  - label: mongo_checkpoint
    memory:
      compaction_interval: "" # disable compaction/expiry (token persists for process lifetime)

input:
  label: "mongo_cdc"
  mongodb_cdc:
    url:      "mongodb://cdc_user:secret@localhost:27017/?replicaSet=rs0"
    database: mydb
    collections:
      - orders
      - customers
    checkpoint_cache:    mongo_checkpoint
    checkpoint_key:      mongodb_cdc_checkpoint   # default
    checkpoint_interval: 5s                        # default
    checkpoint_limit:    1000                      # default
    stream_snapshot:     true
    snapshot_parallelism: 2
    read_batch_size: 1000                          # default
    read_max_wait:   1s                            # default
    document_mode:   update_lookup                 # default
    json_marshal_mode: canonical                   # default

pipeline:
  processors:
    - mapping: |
        # Route each event to a per-collection topic
        meta topic = "mongo.cdc." + meta("collection")
        # Tag the key with the document _id.
        # In canonical json_marshal_mode (the default) _id is an ExtJSON
        # object like {"$oid":"..."}, so extract the inner string:
        meta msg_key = this._id."$oid" | this._id.string() | ""

output:
  redpanda:
    seed_brokers: ["localhost:9092"]
    topic:        ${! meta("topic") }
    key:          ${! meta("msg_key") }
    max_in_flight: 256

3. Run the pipeline

rpk connect run mongodb-cdc-pipeline.yaml

4. Verify messages arrive

rpk topic consume mongo.cdc.orders --num 5 --brokers localhost:9092

Prerequisites

• MongoDB 4.0 or higher — the connector checks the server version at startup and returns an error for < 4.0.
• Replica set or sharded cluster — Change Streams are not available on standalone instances.
• User privileges: at minimum the read role on the target database — this covers find (snapshot), listCollections (schema discovery), and the changeStream action (Change Streams). For parallel snapshots via splitVector on self-managed clusters, also grant clusterManager on admin; alternatively set snapshot_auto_bucket_sharding: true to avoid that requirement. The hello/buildInfo startup commands require no special role.
• Redpanda Enterprise license — the mongodb_cdc component is gated by license.CheckRunningEnterprise.

Core Concepts

Change Streams and Resume Tokens

mongodb_cdc opens a MongoDB Change Stream over the target database, filtered to the configured collections. MongoDB tracks position via an opaque resume token (not an integer offset). The connector stores the latest acknowledged token in the configured cache resource every checkpoint_interval (default 5s), writes one final token on clean shutdown, and — as of Connect 4.106.0 — writes a checkpoint as soon as the initial snapshot completes and is fully acknowledged, so a restart after the snapshot resumes the stream instead of re-snapshotting.

On startup, if no cached token is found, the connector:

1. Records the current oplog position.
2. If stream_snapshot: true, reads all existing documents (operation = "read").
3. Opens the change stream starting just after the recorded oplog position.

If a token is found, the connector resumes directly from that position using ResumeAfter.

Document Modes

The document_mode field controls what body is emitted for update and delete events:

Message Metadata

Every message emitted by mongodb_cdc carries:

Schema Detection

The connector uses a two-tier strategy to populate the schema metadata:

1. At startup, it queries each collection's $jsonSchema validator. If found, this provides accurate types and required/optional field classification.
2. When no validator exists, schema is inferred from the first document seen per collection. All fields are marked optional.

The schema is re-inferred when the top-level field set of a document changes. Type changes within existing fields and nested subdocument structural changes are not auto-detected — restart to force a full schema refresh.

Recommendation: for schema-registry targets with compatibility modes, configure a $jsonSchema validator on each watched collection to stabilize the schema.

AWS IAM Authentication (MongoDB Atlas)

Since Connect 4.106.0, mongodb_cdc (and the mongodb input, output, processor and cache) can authenticate with the driver-native MONGODB-AWS mechanism instead of a static username and password, via an aws config block:

input:
  mongodb_cdc:
    url: "mongodb+srv://cluster0.abc123.mongodb.net/"
    database: mydb
    aws:
      enabled: true          # default false
    collections: [orders]
    checkpoint_cache: mongo_checkpoint

Rules that matter when choosing this path:

• The Atlas database user must be created with the AWS IAM authentication type, and connections require TLS.
• aws.enabled: true is mutually exclusive with username/password and with credentials embedded in url — either combination is a startup error.
• With no static keys or roles configured, the ambient AWS credential chain (environment variables, EC2 instance profile, EKS pod role) is used and expiring credentials are refreshed automatically. Prefer this for long-running pipelines.
• Role assumption (aws.role, aws.roles, or session tokens) is rejected for the mongodb processor and cache, which establish their client once at creation and cannot refresh expiring session credentials. Use the ambient chain or long-lived access keys there.
• With role assumption on mongodb_cdc, credentials are re-resolved after the initial snapshot completes, so streaming starts with a full session — but the snapshot itself must finish inside one session, because snapshot progress is not checkpointed. For very large snapshots, prefer the ambient chain.

Per-field detail (region, session_duration, id/secret/token, role/role_external_id, roles[]) is in Config Reference and the generated connector reference.

Snapshot Phase

When stream_snapshot: true and no resume token is cached, the connector snapshots all documents in each collection before streaming live changes.

• Per-collection concurrency (always): each collection is snapshotted in its own goroutine, so multiple collections are always snapshotted concurrently regardless of snapshot_parallelism.
• Within-collection cursor scan (snapshot_parallelism: 1, the default): each collection is read sequentially with a single cursor.
• Within-collection parallel (snapshot_parallelism > 1): each collection is split into snapshot_parallelism _id-range buckets read concurrently, using splitVector (self-managed, requires clusterManager role) or $bucketAuto (when snapshot_auto_bucket_sharding: true, for Atlas where splitVector is disallowed).

The read_batch_size field controls the MongoDB cursor batch size for both snapshot and streaming phases.

Checkpoint Cache

The checkpoint_cache field is required — you must provide a named cache resource. This cache stores the resume token as BSON Extended JSON.

For development (non-persistent, resets on restart):

cache_resources:
  - label: mongo_checkpoint
    memory:
      compaction_interval: "" # empty string disables compaction and expiry

For production (persistent across restarts), use Redis:

cache_resources:
  - label: mongo_checkpoint
    redis:
      url: redis://localhost:6379

Enabling Pre and Post Images (MongoDB 6+)

To use document_mode: pre_and_post_images, you must enable this feature on each collection:

// Enable at collection level
db.runCommand({
  collMod: "orders",
  changeStreamPreAndPostImages: { enabled: true }
})

This requires MongoDB 6.0+ — changeStreamPreAndPostImages was introduced in 6.0. Use document_mode: update_lookup on MongoDB 4.x–5.x.

Operational Notes

Oplog Window

The resume token references a position in the oplog. If the pipeline falls behind or is stopped for longer than the oplog retention window (default 24 hours on Atlas; configurable with --oplogMinRetentionHours on self-managed clusters), MongoDB can no longer resume from the stored token. Size the oplog so the window comfortably exceeds the longest expected lag or downtime under peak write load.

Unresumable Position Recovery

When a stored position can no longer be resumed from (aged out of the oplog, or another non-resumable change-stream condition), the connector no longer just errors out — it recovers rather than retrying a dead position:

• With stream_snapshot: true — it clears the checkpoint and re-runs the snapshot, which loses nothing. A breaker bounds this: after 3 consecutive recoveries with no change-stream advance in between (usually a snapshot that takes longer than the oplog window), it keeps the checkpoint and fails on every reconnect with an actionable error instead of re-snapshotting forever. Fix the underlying cause — grow the oplog, speed the snapshot up — then restart the pipeline (or delete the checkpoint cache entry) to resume recovery.
• With stream_snapshot: false — there is no snapshot to re-run, so recovery would mean skipping the changes between the lost position and now. The connector refuses to do that silently: on_unresumable_position (default fail) preserves the checkpoint and errors on every reconnect; skipping the gap must be opted into with on_unresumable_position: reset.

checkpoint_write_timeout (default 10s, advanced) bounds the two checkpoint writes that run outside the read loop — storing the position a completed snapshot reached, and clearing an unresumable one — so a slow cache cannot extend shutdown indefinitely. Raise it for slow remote caches (redis, dynamodb), where losing the post-snapshot write costs a full re-snapshot on the next start.

Restart Behavior

On restart with a valid cached token, the connector immediately opens the change stream at the stored position — no snapshot is performed regardless of the stream_snapshot setting. The snapshot only runs when no checkpoint exists; because the completed snapshot's position is now checkpointed once fully acknowledged, a restart after a finished snapshot resumes the stream rather than snapshotting again.

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. A change stream is one totally-ordered cursor over the replica set's oplog, so throughput does not grow with more collections, and the input does not use CPU beyond roughly two cores. This is a property of MongoDB change streams, not of the connector. To scale past that ceiling, shard the cluster: a change stream against a sharded cluster merges parallel per-shard cursors server-side.

Atlas / Restricted Environments

On MongoDB Atlas, the splitVector command is not available. Set snapshot_auto_bucket_sharding: true to use $bucketAuto aggregation for parallel snapshots instead.

Enterprise Features Around the CDC Topic

The mongodb_cdc input is itself an Enterprise connector, and the topics it feeds pair with other Redpanda enterprise differentiators (all require a valid Enterprise license — see Enterprise Integration for nested keys and examples):

• Iceberg Topics — make the CDC output topic a queryable Iceberg table. Set cluster iceberg_enabled=true, then topic redpanda.iceberg.mode (key_value for raw CDC JSON, or value_schema_id_prefix/value_schema_latest for schema-serialized events), plus redpanda.iceberg.delete, redpanda.iceberg.invalid.record.action (drop/dlq_table), redpanda.iceberg.partition.spec, redpanda.iceberg.target.lag.ms.
• Server-side Schema ID Validation — cluster enable_schema_id_validation (none/redpanda/compat); topic redpanda.value.schema.id.validation + redpanda.value.subject.name.strategy (and key equivalents).
• Tiered Storage — long-term CDC retention: topic redpanda.remote.write + redpanda.remote.read, with retention.local.target.ms/.bytes bounding the local footprint.
• Sink security — TLS (tls.client_certs[], tls.root_cas_file) and SASL (sasl[].mechanism: SCRAM-SHA-512/PLAIN/OAUTHBEARER); OAUTHBEARER/OIDC and Kerberos broker auth are Enterprise.
• RBAC, Connect secrets management, FIPS, allow/deny component lists — harden the pipeline and resolve the MongoDB/SASL passwords from a remote secret store.

Reference Directory

• Config Reference: Every mongodb_cdc input field with type, default, and description grounded in source.
• Setup MongoDB: Replica-set/sharded-cluster requirement, user privileges, Atlas specifics, and oplog configuration.
• Pipeline and Output: Full pipeline YAML examples, message/metadata shape, per-collection routing, cache options, and restart/resume semantics.
• Enterprise Integration: Enterprise features around the CDC topic and their nested config keys — Iceberg Topics (redpanda.iceberg.*), Server-side Schema ID Validation (enable_schema_id_validation, redpanda.{key,value}.schema.id.validation), Tiered Storage (redpanda.remote.*, retention.local.target.*), sink TLS + SASL (SCRAM/OAUTHBEARER/Kerberos), RBAC, Connect secrets management, FIPS, and allow/deny lists. Notes which require an Enterprise license.