connect-cdc-mongodb

---
name: connect-cdc-mongodb
description: >-
  Streams change data capture from MongoDB into Redpanda or Kafka using Redpanda
  Connect's mongodb_cdc input, built on MongoDB Change Streams over a replica set or
  sharded cluster. Use when configuring mongodb_cdc, setting up Change Streams,
  enabling an initial snapshot before live streaming, choosing between update_lookup
  and pre_and_post_images document modes, or checkpointing with resume tokens in a
  cache resource. Also covers Redpanda Enterprise destination features like Iceberg
  Topics, Tiered Storage, and server-side Schema ID Validation, all of which require a
  Redpanda Enterprise license.
---

# 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.

```bash
# 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)

```yaml
# 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

```bash
rpk connect run mongodb-cdc-pipeline.yaml
```

### 4. Verify messages arrive

```bash
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:

| Mode | Updates | Deletes |
|------|---------|---------|
| `update_lookup` (default) | Full document after the update (via `UpdateLookup`). Falls back to `documentKey` if the document was deleted before lookup. | Only `_id` populated (documentKey). |
| `pre_and_post_images` | Full document before and after (requires MongoDB 6.0+ and `changeStreamPreAndPostImages` enabled on each collection). | Full document before deletion. |
| `partial_update` | A structured diff: `{_id, operations: [{path, type, value}]}` where `type` is `set`, `unset`, or `truncatedArray`. Enables `showExpandedEvents` on MongoDB 6.1+. | Only `documentKey`. |

### Message Metadata

Every message emitted by `mongodb_cdc` carries:

| Metadata key | Value |
|---|---|
| `operation` | `"read"` (snapshot), `"insert"`, `"update"`, `"replace"`, or `"delete"` |
| `collection` | Collection name (e.g., `"orders"`) |
| `operation_time` | BSON timestamp in JSON form: `{"$timestamp":{"t":<unix_sec>,"i":<ordinal>}}` |
| `schema` | Inferred or validator-derived schema in benthos common schema format (immutable; absent when no schema can be determined, e.g., deletes without pre-images) |

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

```yaml
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](references/config-reference.md) 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):

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

For production (persistent across restarts), use Redis:

```yaml
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:

```javascript
// 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](references/enterprise-integration.md)
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](references/config-reference.md): Every `mongodb_cdc` input field with type, default, and description grounded in source.
- [Setup MongoDB](references/setup-mongodb.md): Replica-set/sharded-cluster requirement, user privileges, Atlas specifics, and oplog configuration.
- [Pipeline and Output](references/pipeline-and-output.md): Full pipeline YAML examples, message/metadata shape, per-collection routing, cache options, and restart/resume semantics.
- [Enterprise Integration](references/enterprise-integration.md): 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.

• SKILL.md 17 KB
• references/SOURCES.md 10 KB
• references/config-reference.md 16 KB
• references/enterprise-integration.md 11 KB
• references/pipeline-and-output.md 12 KB
• references/setup-mongodb.md 9 KB