> ## Documentation Index
> Fetch the complete documentation index at: https://docs.startree.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# MongoDB CDC Ingestion

> Consume MongoDB Change Streams directly into a realtime Pinot table.

MongoDB CDC ingestion lets a StarTree Pinot realtime table consume MongoDB Change Streams directly, without first copying changes through Kafka, Kinesis, Pulsar, or another durable stream.

<Info>
  This connector is intended for advanced realtime ingestion designs where the MongoDB source namespaces, partitioning, credentials, and replay window are explicitly managed. If you only need to consume Debezium-formatted MongoDB events from Kafka, use the [Debezium decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/debezium) instead.
</Info>

## How It Works

The connector maps each configured MongoDB source partition to one Pinot partition group:

```text theme={null}
app.orders_00 -> Pinot partition group 0
app.orders_01 -> Pinot partition group 1
app.orders_02 -> Pinot partition group 2
```

Use the StarTree MongoDB CDC consumer factory:

```text theme={null}
ai.startree.pinot.plugin.stream.mongodb.MongoCdcConsumerFactory
```

Pinot controllers resolve source-partition metadata and checkpoints. Pinot servers open and resume MongoDB Change Stream cursors for the partitions they own. Pinot replication creates another cursor per partition replica, so plan source-side Change Stream load as:

```text theme={null}
source partitions x Pinot replicas
```

For sharded MongoDB deployments, a cursor opened through `mongos` can fan out to every shard. Plan closer to:

```text theme={null}
source partitions x Pinot replicas x MongoDB shards
```

## Source-Partition Contract

MongoDB Change Streams do not expose Kafka-style durable partitions. The connector therefore requires a manifest of physical, disjoint MongoDB namespaces.

Follow these rules:

* Partition IDs are dense, stable integers: `0`, `1`, `2`, and so on.
* Every descriptor has a globally unique and immutable `sourceId`, such as `atlas-prod-a/app/orders-00`.
* `_id` values must be globally unique across the configured namespaces.
* Every insert, update, replace, and delete for one primary key must stay in the same source partition for that key's lifetime.
* Do not remove, renumber, rebind, or repartition an active source partition. Create a new Pinot table, snapshot/backfill, catch up CDC, and cut over queries instead.
* Do not use `hash(_id) % currentPartitionCount` for upstream routing. Increasing the partition count would move existing keys and break upsert partition affinity.

A common routing pattern is to encode the source partition into the immutable `_id`:

```text theme={null}
_id = "02:ord_123" -> app.orders_02
_id = "07:ord_456" -> app.orders_07
```

New partitions can receive only newly created IDs. Existing IDs must continue to route to their original namespace.

## Prerequisites

* MongoDB Atlas, a replica set, or a sharded cluster. Standalone `mongod` processes do not support Change Streams.
* MongoDB credentials available to Pinot controllers and servers.
* `find` and `changeStream` permissions on every source namespace.
* If you use registry mode, `find` on the manifest collection and `listCollections` on source databases.
* If you use `outputMode=full_document`, MongoDB Server 6.0 or later with Change Stream pre/post-images enabled.
* An oplog and post-image retention window longer than the maximum Pinot outage and recovery window.

## Static Manifest Mode

Use static mode when the source partition list is fixed. Omit `partitionDiscoveryMode` and set `sourcePartitions` to a JSON array encoded as a string.

```json theme={null}
{
  "streamType": "mongodb",
  "stream.mongodb.topic.name": "mongo-orders",
  "stream.mongodb.consumer.factory.class.name": "ai.startree.pinot.plugin.stream.mongodb.MongoCdcConsumerFactory",
  "stream.mongodb.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
  "stream.mongodb.consumer.prop.auto.offset.reset": "largest",
  "sourcePartitions": "[{\"partitionId\":0,\"epoch\":1,\"sourceId\":\"atlas-prod-a/app/orders-00\",\"connectionStringEnvironmentVariable\":\"MONGODB_ATLAS_URI\",\"database\":\"<DATABASE>\",\"collection\":\"<COLLECTION>\"},{\"partitionId\":1,\"epoch\":1,\"sourceId\":\"atlas-prod-a/app/orders-01\",\"connectionStringEnvironmentVariable\":\"MONGODB_ATLAS_URI\",\"database\":\"<DATABASE>\",\"collection\":\"<COLLECTION>\"}]",
  "fullDocument": "required",
  "outputMode": "full_document",
  "cursorBatchSize": "100",
  "maxBatchMessages": "500",
  "maxBatchBytes": "4194304",
  "maxAwaitTimeMs": "500",
  "stream.mongodb.idle.timeout.millis": "180000"
}
```

In this mode, every descriptor requires:

| Field                                                                         | Description                                                                                                            |
| ----------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------- |
| `partitionId`                                                                 | Dense Pinot partition ID.                                                                                              |
| `epoch`                                                                       | Positive operator-controlled lineage epoch. Treat it as immutable for an active table.                                 |
| `sourceId`                                                                    | Globally unique source identity for the physical namespace lineage.                                                    |
| `database`                                                                    | MongoDB database name.                                                                                                 |
| `collection`                                                                  | MongoDB collection name.                                                                                               |
| `connectionRef`, `connectionString`, or `connectionStringEnvironmentVariable` | Exactly one connection source. Prefer environment variables or connection refs for production.                         |
| `initialOffset`                                                               | Required when appending capacity to an existing table. Optional for initial creation with `auto.offset.reset=largest`. |

## Registry Manifest Mode

Use `partitionDiscoveryMode=mongodb_manifest` when an upstream routing service can append new physical bucket collections. The table config allowlists connection references and points Pinot at one authoritative registry document.

```json theme={null}
{
  "streamType": "mongodb",
  "stream.mongodb.topic.name": "mongo-orders",
  "stream.mongodb.consumer.factory.class.name": "ai.startree.pinot.plugin.stream.mongodb.MongoCdcConsumerFactory",
  "stream.mongodb.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
  "stream.mongodb.consumer.prop.auto.offset.reset": "largest",
  "partitionDiscoveryMode": "mongodb_manifest",
  "connectionRefs": "{\"atlas-prod-a\":{\"connectionStringEnvironmentVariable\":\"MONGODB_ATLAS_URI\"}}",
  "partitionManifest.connectionRef": "atlas-prod-a",
  "partitionManifest.database": "pinot_cdc_control",
  "partitionManifest.collection": "partition_manifests",
  "partitionManifest.id": "mongo-orders",
  "maxPartitionCount": "256",
  "fullDocument": "required",
  "outputMode": "full_document",
  "cursorBatchSize": "100",
  "maxBatchMessages": "500",
  "maxBatchBytes": "4194304",
  "maxAwaitTimeMs": "500"
}
```

Example registry document:

```json theme={null}
{
  "_id": "mongo-orders",
  "schemaVersion": 1,
  "generation": {"$numberLong": "3"},
  "partitions": [
    {
      "partitionId": 0,
      "epoch": {"$numberLong": "1"},
      "sourceId": "atlas-prod-a/app/orders-00",
      "connectionRef": "atlas-prod-a",
      "database": "<DATABASE>",
      "collection": "<COLLECTION>",
      "collectionUuid": {"$binary": {"base64": "AAAAAAAAAAAAAAAAAAAAAA==", "subType": "04"}},
      "initialOffset": {
        "resumeToken": {"_data": "<opaque MongoDB resume token>"},
        "logicalTime": {"$timestamp": {"t": 1810000000, "i": 1}}
      }
    }
  ]
}
```

Registry mode fails closed if an existing descriptor changes, a partition is removed, a partition ID is skipped, a namespace UUID changes, or a later generation mutates an old prefix. `maxPartitionCount` is a safety limit that prevents a malformed registry from creating unbounded Pinot segments or MongoDB cursors.

## Output Modes

| Mode            | Use Case                                     | Notes                                                                                                                                              |
| --------------- | -------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
| `change_stream` | Preserve the MongoDB Change Stream envelope. | Default. Leaves `fullDocument` nested and defaults to `fullDocument=default`.                                                                      |
| `full_document` | Maintain a current-state Pinot upsert table. | Emits document fields at the top level and adds reserved `__mongodb_cdc_*` metadata. Requires `fullDocument=required` and MongoDB pre/post-images. |

For `full_document`, enable post-images on every source collection:

```javascript theme={null}
db.runCommand({
  collMod: "orders_00",
  changeStreamPreAndPostImages: {enabled: true}
})
```

Source documents must not contain connector-reserved fields such as:

```text theme={null}
__mongodb_cdc_document_id
__mongodb_cdc_document_key
__mongodb_cdc_operation_type
__mongodb_cdc_is_deleted
__mongodb_cdc_event_time_ms
__mongodb_cdc_comparison_value
__mongodb_cdc_resume_token
__mongodb_cdc_partition_id
__mongodb_cdc_partition_epoch
```

## Current-State Upsert Example

Schema:

```json theme={null}
{
  "schemaName": "mongo_orders",
  "primaryKeyColumns": ["__mongodb_cdc_document_id"],
  "dimensionFieldSpecs": [
    {"name": "__mongodb_cdc_document_id", "dataType": "STRING"},
    {"name": "__mongodb_cdc_document_key", "dataType": "STRING"},
    {"name": "__mongodb_cdc_comparison_value", "dataType": "STRING"},
    {"name": "__mongodb_cdc_operation_type", "dataType": "STRING"},
    {"name": "__mongodb_cdc_is_deleted", "dataType": "BOOLEAN"},
    {"name": "__mongodb_cdc_partition_id", "dataType": "INT"},
    {"name": "__mongodb_cdc_partition_epoch", "dataType": "LONG"},
    {"name": "status", "dataType": "STRING"},
    {"name": "region", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "amount", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "__mongodb_cdc_event_time_ms",
      "dataType": "LONG",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
```

Realtime upsert config:

```json theme={null}
{
  "upsertConfig": {
    "mode": "FULL",
    "comparisonColumns": ["__mongodb_cdc_comparison_value"],
    "deleteRecordColumn": "__mongodb_cdc_is_deleted",
    "snapshot": "ENABLE",
    "preload": "ENABLE"
  },
  "routing": {
    "instanceSelectorType": "strictReplicaGroup"
  }
}
```

`comparisonColumns` uses the connector-generated comparison value so MongoDB event order is preserved. `strictReplicaGroup` is required so a query sees a consistent upsert view for each partition.

## Capturing Initial Offsets

When creating a table with `auto.offset.reset=largest`, the controller captures a starting boundary for each partition that does not already provide `initialOffset`.

To include historical documents:

1. Capture an exact resume boundary for each MongoDB namespace.
2. Snapshot/backfill the historical data up to that boundary.
3. Configure each descriptor with the captured `initialOffset`.
4. Start realtime CDC consumption from those offsets.

Descriptor example:

```json theme={null}
{
  "partitionId": 2,
  "epoch": 1,
  "sourceId": "atlas-prod-a/app/orders-02",
  "connectionStringEnvironmentVariable": "MONGODB_ATLAS_URI",
  "database": "<DATABASE>",
  "collection": "<COLLECTION>",
  "initialOffset": {
    "resumeToken": {"_data": "<opaque MongoDB resume token>"},
    "logicalTime": {"$timestamp": {"t": 1810000000, "i": 1}}
  }
}
```

Do not substitute a bare `clusterTime`; the offset must include the resume token and its safe logical time.

## Adding Capacity

To append a partition safely:

1. Create a new, quiescent physical namespace.
2. Capture the namespace UUID and exact initial offset.
3. Append only the next highest dense `partitionId`.
4. Wait until Pinot reports the new partition group consuming on the required replicas.
5. Route only new keys to the new namespace.

Do not configure a primary-key `segmentPartitionConfig` on a table whose MongoDB source-partition count can grow. A growing partition count changes key ownership for older rows.

## Troubleshooting

| Symptom                                 | Likely Cause                                                                | Fix                                                                                                     |
| --------------------------------------- | --------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------- |
| Connector fails before opening a cursor | Descriptor identity or fingerprint mismatch.                                | Check `sourceId`, `epoch`, database, collection, UUID, output mode, and watch options.                  |
| Resume fails after an outage            | MongoDB oplog or post-image history expired.                                | Create a new table from a snapshot and matching Change Stream boundaries.                               |
| Deletes are not reflected in Pinot      | `full_document` mode is not configured with delete handling.                | Use the connector-generated `__mongodb_cdc_is_deleted` as `deleteRecordColumn`.                         |
| Source load is higher than expected     | Cursor count scales with partitions, Pinot replicas, and MongoDB shards.    | Reduce partitions or replicas, or isolate source partitions on separate replica sets/Atlas deployments. |
| New partitions never start consuming    | Registry generation is invalid or descriptor lacks an exact initial offset. | Ensure the registry appends only dense new partitions and includes `initialOffset`.                     |