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. If you have not chosen an ingestion path yet, start at Connect MongoDB, which compares this connector against the Kafka and mongodump routes.

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

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

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

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:

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:

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

Tuning Options and Defaults

Rejected Options

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

​Tuning Options and Defaults

​Rejected Options

​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

Tuning Options and Defaults

Rejected Options

Current-State Upsert Example

Capturing Initial Offsets

Adding Capacity

Troubleshooting