MongoDB CDC Ingestion

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.

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

How It Works

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

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:

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:

source partitions x Pinot replicas

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

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:

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

{
  "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.

{
  "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:

{
  "_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:

db.runCommand({
  collMod: "orders_00",
  changeStreamPreAndPostImages: {enabled: true}
})

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

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

{
  "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:

{
  "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:

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

Descriptor example:

{
  "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:

Create a new, quiescent physical namespace.
Capture the namespace UUID and exact initial offset.
Append only the next highest dense partitionId.
Wait until Pinot reports the new partition group consuming on the required replicas.
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`.

Get Started

Ingestion

External Tables

Query Data

Manage Data

Observability

Visualize Data

Cluster Operations

Manage Security

Release Notes

Reference

How It Works

Source-Partition Contract

Prerequisites

Static Manifest Mode

Registry Manifest Mode

Output Modes

Current-State Upsert Example

Capturing Initial Offsets

Adding Capacity

Troubleshooting

​How It Works

​Source-Partition Contract

​Prerequisites

​Static Manifest Mode

​Registry Manifest Mode

​Output Modes

​Current-State Upsert Example

​Capturing Initial Offsets

​Adding Capacity

​Troubleshooting

How It Works

Source-Partition Contract

Prerequisites

Static Manifest Mode

Registry Manifest Mode

Output Modes

Current-State Upsert Example

Capturing Initial Offsets

Adding Capacity

Troubleshooting