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

If you have not chosen an ingestion path yet, start at [Connect MongoDB](/corecapabilities/ingestdata/mongodb), which compares this connector against the Kafka and `mongodump` routes.

<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 any of the connector's reserved fields:

```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_cluster_time_seconds
__mongodb_cdc_cluster_time_increment
__mongodb_cdc_comparison_value
__mongodb_cdc_resume_token
__mongodb_cdc_partition_id
__mongodb_cdc_partition_epoch
```

## Tuning Options and Defaults

Every option below is optional. The values shown in the examples on this page are the defaults unless noted.

| Option | Default | Notes |
| - | - | - |
| `cursorBatchSize` | `100` | MongoDB Change Stream cursor batch size. |
| `maxBatchMessages` | `500` | Maximum events returned to Pinot per fetch. |
| `maxBatchBytes` | `4194304` (4 MiB) | Maximum bytes returned per fetch. |
| `maxAwaitTimeMs` | `500` | Must be strictly less than Pinot's stream fetch timeout. MongoDB server-selection, connect, socket-read, and operation timeouts are all capped below this budget. |
| `maxPartitionCount` | `1024` | Registry mode only. A safety bound, not a target — it caps how many Pinot segments and MongoDB cursors a malformed or compromised registry can create. |

Size the MongoDB URI's `maxPoolSize` for the maximum number of MongoDB CDC partitions that can be active in one Pinot JVM across all tables, plus controller metadata requests. Partitions sharing an identical connection string and timeout settings share one client and pool; distinct credentials, endpoints, or timeouts get their own.

### Rejected Options

The connector fails closed rather than accept these:

| Option | Why |
| - | - |
| `pipeline` | A generic source predicate cannot prove exhaustive, non-overlapping primary-key routing, and can hide deletes or lifecycle events. A top-level `pipeline` is rejected as a legacy singleton key, and the error suggests moving it into each `sourcePartitions` descriptor — but descriptors do not accept it either. There is no supported place for it. |
| `showExpandedEvents=true` | Expanded DDL and split large events need a broader checkpoint and reassembly contract. |
| Pinot realtime offset auto-reset | The connector never silently skips to latest. A lost resume position is a permanent failure, not something to paper over. |
| `fullDocument=updateLookup` or `whenAvailable` | Rejected in every output mode, not only `full_document`: `fullDocument` must be `default` or `required` so replayed rows stay replica-deterministic. Independent Pinot replicas could otherwise observe different post-images or nullability during replay. |

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

For step 1, the connector JAR exposes a bootstrap API that prints the descriptor-ready Extended JSON. Run it from a short Java program with the shaded connector JAR on the classpath, using the same `fullDocument` and `maxAwaitTimeMs` values as the target table:

```java theme={null}
import ai.startree.pinot.plugin.stream.mongodb.MongoCdcInitialOffsetCapture;
import ai.startree.pinot.plugin.stream.mongodb.MongoCdcCollectionMetadata;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.model.changestream.FullDocument;
import org.bson.BsonDocument;

try (MongoClient client = MongoClients.create(System.getenv("MONGODB_ATLAS_URI"))) {
  System.out.println(new BsonDocument("collectionUuid",
      MongoCdcCollectionMetadata.getCollectionUuid(client, "<DATABASE>", "<COLLECTION>", 5_000)).toJson());
  BsonDocument initialOffset = MongoCdcInitialOffsetCapture.capture(
      client, "<DATABASE>", "<COLLECTION>", FullDocument.REQUIRED, 500, 5_000);
  System.out.println(initialOffset.toJson());
}
```

Capture each namespace independently, paste the printed document in as that descriptor's `initialOffset`, then snapshot that namespace. Registry mode also needs the printed `collectionUuid`.

Boundary capture is per namespace and is not a transactionally atomic cut across collections.

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


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.