# MongoDB CDC Patterns

> For the complete documentation index, see [llms.txt](https://docs.redpanda.com/llms.txt). Component-specific: [connect-full.txt](https://docs.redpanda.com/connect-full.txt)

---
title: MongoDB CDC Patterns
latest-connect-version: 4.104.0
latest-operator-version: v26.2.1
latest-console-tag: v3.10.0
latest-redpanda-tag: v26.2.1
docname: mongodb_cdc
page-component-name: connect
page-version: master
page-component-version: master
page-component-title: Connect
page-relative-src-path: mongodb_cdc.adoc
page-edit-url: https://github.com/redpanda-data/rp-connect-docs/edit/main/modules/cookbooks/pages/mongodb_cdc.adoc
description: Learn how to capture, filter, transform, and route MongoDB change data capture (CDC) events with Redpanda Connect.
page-topic-type: cookbook
personas: streaming_developer, data_engineer
learning-objective-1: Apply reusable patterns for capturing MongoDB CDC events
learning-objective-2: Adapt patterns to route CDC data to Redpanda, S3, and other destinations
learning-objective-3: Identify patterns for filtering and transforming change events
page-git-created-date: "2026-06-17"
page-git-modified-date: "2026-06-17"
---

<!-- Source: https://docs.redpanda.com/connect/cookbooks/mongodb_cdc.md -->

The `mongodb_cdc` input captures document-level changes from MongoDB collections using change streams. Use these patterns to filter, transform, and route MongoDB CDC events to Redpanda, S3, and other destinations.

Use this cookbook to:

-   Apply reusable patterns for capturing MongoDB CDC events

-   Adapt patterns to route CDC data to Redpanda, S3, and other destinations

-   Identify patterns for filtering and transforming change events


## [](#prerequisites)Prerequisites

Before using these patterns, configure the following.

### [](#redpanda-cli)Redpanda CLI

Install the Redpanda CLI (`rpk`) to run Redpanda Connect. See [rpk installation](https://docs.redpanda.com/connect/get-started/quickstarts/rpk/) for installation instructions.

### [](#mongodb-replica-set)MongoDB replica set

The `mongodb_cdc` input requires a MongoDB database running as a replica set or in a sharded cluster. Change streams are not available on standalone MongoDB instances.

To verify your deployment is running as a replica set:

```javascript
rs.status()
```

For cloud-managed MongoDB, [MongoDB Atlas](https://www.mongodb.com/docs/atlas/) uses replica sets by default.

### [](#checkpoint-cache)Checkpoint cache

The `mongodb_cdc` input requires a [cache](https://docs.redpanda.com/connect/components/caches/about/) to store the oplog position between restarts. The examples in this cookbook use a Redis cache:

```bash
export REDIS_URL=redis://localhost:6379
```

For production use, any persistent cache supported by Redpanda Connect works, such as `aws_dynamodb` or `postgres`.

### [](#environment-variables)Environment variables

The examples in this cookbook use environment variables for configuration:

```bash
export MONGODB_URL="mongodb://user:password@localhost:27017" (1)
export MONGODB_DB="mydb" (2)
export REDIS_URL=redis://localhost:6379 (3)
export REDPANDA_BROKERS=localhost:9092 (4)
export S3_BUCKET=cdc-archive (5)
```

| 1 | The MongoDB connection string. |
| --- | --- |
| 2 | The name of the database to capture changes from. |
| 3 | The Redis URL for checkpoint storage. |
| 4 | The Redpanda broker addresses (for Redpanda output examples). |
| 5 | The S3 bucket name (for S3 output examples). |

## [](#capture-cdc-events)Capture CDC events

The simplest pattern captures all change events from MongoDB collections and outputs them with metadata:

```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.collection = meta("collection")
        root.operation_time = meta("operation_time")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

For details on the CDC event message structure and available metadata fields, see the [metadata](https://docs.redpanda.com/connect/components/inputs/mongodb_cdc/#_metadata) section in the connector reference.

## [](#filter-cdc-events)Filter CDC events

Filter events to process only specific change types using the `operation` metadata field:

```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: false

pipeline:
  processors:
    - mapping: |
        root = if meta("operation") == "delete" || meta("operation") == "read" {
          deleted()
        }
    - mapping: |
        root.operation = meta("operation")
        root.collection = meta("collection")
        root.data = this
        root.timestamp = now()

output:
  stdout:
    codec: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

This pattern:

-   Filters to only `insert`, `update`, and `replace` operations, dropping `delete` and `read` events

-   Transforms the event to a simplified format with a timestamp


## [](#route-to-redpanda)Route to Redpanda

Stream MongoDB changes to Redpanda for real-time processing:

```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
      - customers
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        meta topic = meta("collection")

output:
  redpanda:
    seed_brokers:
      - ${REDPANDA_BROKERS}
    topic: ${! meta("topic") }
    key: ${! json("_id") }
    batching:
      count: 100
      period: 1s

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

This pattern:

-   Uses the collection name as the Redpanda topic

-   Batches messages for efficient delivery

-   Sets the message key to the document’s `_id` field


## [](#route-to-s3)Route to S3

Archive CDC events to Amazon S3 for long-term storage and analytics:

```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true

pipeline:
  processors:
    - mapping: |
        root.operation = meta("operation")
        root.collection = meta("collection")
        root.data = this
        root.timestamp = now()

output:
  aws_s3:
    bucket: ${S3_BUCKET}
    path: >-
      cdc/${! meta("collection") }/${! timestamp_unix().format_timestamp("2006/01/02/15") }/${! uuid_v4() }.ndjson
    batching:
      count: 1000
      period: 5m
      processors:
        - archive:
            format: lines

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

This pattern:

-   Organizes files by collection and time-based partitions (year/month/day/hour)

-   Batches events and archives them as newline-delimited JSON

-   Uses UUID file names to prevent collisions


## [](#route-by-event-type)Route by event type

Route different event types to different destinations:

```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: false

output:
  switch:
    cases:
      - check: meta("operation") == "insert"
        output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: mongodb.orders.inserts
      - output:
          redpanda:
            seed_brokers:
              - ${REDPANDA_BROKERS}
            topic: mongodb.orders.changes

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

This pattern:

-   Routes `insert` events to one Redpanda topic and all other change events to another

-   Supports specialized downstream consumers per operation type


## [](#configure-replication-mode)Configure replication mode

The `mongodb_cdc` input supports two replication modes controlled by the `stream_snapshot` field:

-   `stream_snapshot: true`: Captures a full snapshot of existing collection data before streaming live changes. Use this when you need a complete initial load.

-   `stream_snapshot: false`: Skips the snapshot and streams only changes from the current oplog position. Use this when you only need new changes going forward.


```yaml
input:
  mongodb_cdc:
    url: ${MONGODB_URL}
    database: ${MONGODB_DB}
    collections:
      - orders
    checkpoint_cache: redis_cache
    stream_snapshot: true
    snapshot_parallelism: 4 (1)

cache_resources:
  - label: redis_cache
    redis:
      url: ${REDIS_URL}
```

| 1 | Number of parallel connections to use when reading the snapshot. |
| --- | --- |

> 📝 **NOTE**
>
> If the pipeline restarts during a snapshot, Redpanda Connect must restart the snapshot from the beginning to record a fresh oplog position in the checkpoint cache.

## [](#troubleshoot-common-issues)Troubleshoot common issues

Use these steps to diagnose and fix the most common problems with the `mongodb_cdc` input.

### [](#no-events-received)No events received

If no events arrive:

1.  Verify the MongoDB deployment is a replica set:

    ```javascript
    rs.status()
    ```

2.  Grant the required privileges to the user:

    ```javascript
    db.grantRolesToUser("your_user", [
      { role: "read", db: "mydb" },
      { role: "clusterMonitor", db: "admin" }
    ])
    ```

3.  Check that the collection names in `collections` exist in the specified `database`.


### [](#pipeline-restarts-lose-position)Pipeline restarts lose position

If the pipeline restarts and replays events from the beginning:

-   Verify the checkpoint cache is persistent and accessible.

-   Check that `checkpoint_key` is consistent across restarts (default: `mongodb_cdc_checkpoint`).

-   Use a durable cache backend such as Redis with persistence enabled, `aws_dynamodb`, or `postgres`.


### [](#duplicate-events)Duplicate events

The `mongodb_cdc` input provides at-least-once delivery. If the pipeline fails between checkpoints, events may be re-read on restart. To handle duplicates:

-   Use idempotent processing in downstream systems.

-   Deduplicate using the `operation_time` metadata field.

-   Lower `checkpoint_limit` to reduce the window of possible duplicates.


## [](#next-steps)Next steps

-   [MongoDB CDC input reference](https://docs.redpanda.com/connect/components/inputs/mongodb_cdc/)

-   [Redis cache](https://docs.redpanda.com/connect/components/caches/redis/)

-   [Redpanda output](https://docs.redpanda.com/connect/components/outputs/redpanda/)


## Suggested labs

-   [Stream Jira Issues to Redpanda for Real-Time Metrics](https://docs.redpanda.com/labs/docker-compose/jira-metrics-pipeline/)

[Search all labs](https://docs.redpanda.com/labs)