MongoDB Change Streams: Real-Time Ingest from Document Stores

mongodb change streams are the official cursor API layered on top of the replica-set oplog, and they are the reason "how do you do CDC out of Mongo?" stopped being a research project sometime around 2018 and started being a checklist. Every senior data engineer inheriting a document-store ingest pipeline runs into the same triangle of questions: what does the change event actually look like, how do I survive a consumer restart without dropping events, and how much extra latency do I pay when the downstream sink needs the full document instead of the raw delta the oplog serialised in the first place. Miss any one of those axes and the pipeline either loses data at restart, floods your Mongo primary with updateLookup round-trips, or ships stale rows to a warehouse that the analytics team quietly stops trusting.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "explain the difference between resumeAfter and startAtOperationTime on a change stream cursor" or "why does fullDocument: 'updateLookup' cost a round-trip and when do you use 'required' instead?" or "walk me through what a mongo cdc pipeline looks like when the source is a sharded cluster and the sink is Kafka via the mongo debezium connector." It walks through why mongo change streams replaced the ad-hoc mongo oplog tailers of the pre-3.6 era, the change-event schema every consumer must parse, resume-token discipline for at-least-once semantics, the full-document lookup cost model, the sharded and cluster-wide watch() semantics coordinated through mongos, and the Debezium-versus-native-driver decision that senior engineers ship into every document store cdc project. Each section pairs a teaching block with a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the streaming practice library →, rehearse on the ETL practice library →, and sharpen the tuning axis with the optimization practice library →.

---

On this page

• Why change streams became the default Mongo CDC path
• Change event schema + operationType
• Resume tokens + at-least-once semantics
• Full document lookup + updateDescription
• Cluster-wide + sharded change streams + Debezium
• Cheat sheet — MongoDB change stream recipes
• Frequently asked questions
• Practice on PipeCode

---

1. Why change streams became the default Mongo CDC path

From tailing the oplog by hand to a first-class cursor — the API that made mongo cdc boring

The one-sentence invariant: before 3.6, doing CDC out of Mongo meant opening a tailable cursor against the local.oplog.rs collection, decoding the operation-log documents yourself, and reimplementing every edge case — filtering, resume, sharded coordination — as application code; change streams collapse that entire pile into a single db.collection.watch(...) call that returns a documented, stable, driver-supported event stream. The oplog is still down there doing the work; change streams are the officially blessed reader on top of it.

The four "must-answer" axes interviewers actually probe.

• Change event schema. Every event has an _id (the resume token), an operationType (insert / update / replace / delete / drop / rename / invalidate), a ns (namespace = db.coll), a documentKey (_id of the target document — plus shard key in a sharded cluster), an updateDescription (for updates: updatedFields + removedFields), an optional fullDocument, and a clusterTime. Knowing the fields by heart is table-stakes.
• Resume token discipline. The _id on every event is an opaque BSON blob that encodes a position in the oplog. Persist it after every downstream ack and pass it back as resumeAfter on restart, and the stream picks up exactly where you left off — as long as the oplog still contains that position. Lose the token or lag behind the oplog window and you get a ChangeStreamHistoryLost error.
• Full document lookup cost. The default event payload for an update is a delta — the fields that changed. Downstream sinks that need the current row (warehouses, search indices, materialised views) must either apply the delta themselves or ask the server to hydrate the row with fullDocument: 'updateLookup' — which costs a round-trip and reads whatever the primary has now, not the state at clusterTime.
• Sharded coordination. In a sharded cluster, change streams open against mongos, which fans the watch out to every shard, merges by cluster time, and returns a globally ordered stream. That coordination has a latency cost and creates a subtle ordering caveat when shards drift.

Why the pre-change-streams era was painful.

• Oplog is a capped collection. local.oplog.rs is a fixed-size ring buffer; tailing it required knowing Mongo-internal formats ({op: 'i' | 'u' | 'd', o, o2, ts, t}) and reconstructing intent from low-level ops.
• Manual everything. Client-side filtering, DIY resume via per-shard Timestamp, no sharded merger, and deletes carried only _id — anything else needed a side cache.

What change streams changed.

• Server-side pipeline. watch([{$match: {...}}, {$project: {...}}]) filters and reshapes events before they hit the wire.
• Documented schema + opaque resume tokens. Stable driver API; consumers store one blob and pass it back.
• Three scopes. client.watch(...) (cluster-wide, 4.0+), db.watch(...), collection.watch(...) — all return the same event shape.
• Full-document knob. fullDocument: 'updateLookup' (3.6+), 'required' / 'whenAvailable' (6.0+) make hydration a config option.

Where 2026 landed.

• Native driver cursors. ~30 lines of Python/Node covers 80% of use cases.
• Debezium Mongo connector. Kafka-Connect-hosted; offsets in a Kafka topic; default for Connect shops.
• Managed CDC. Atlas triggers, AWS DMS, Confluent Cloud — hide the resume-token discipline until it breaks.

What interviewers listen for.

• Naming "oplog is a capped collection" first — senior signal.
• Pairing resume token with oplog retention window as the twin at-least-once constraints — required.
• Distinguishing cluster-wide (client.watch) from collection-scoped (collection.watch) — required.
• Pushing back on "just use Debezium" with the "native cursor is 30 lines" argument for smaller teams — senior signal.

Worked example — tailing the oplog by hand vs a change-stream cursor

Detailed explanation. The textbook before-and-after: an older ingest pipeline tails local.oplog.rs with a tailable cursor and hand-rolls filtering, resume, and delete-hydration. A modern pipeline calls collection.watch(...) and moves on. Show the two side-by-side to make the abstraction win explicit.

• The old code. Open local.oplog.rs with cursor_type=TAILABLE_AWAIT, filter for ns == 'shop.orders', decode op codes, look up deleted documents from a side cache. Roughly 300 lines and every one of them was a footgun.
• The new code. collection.watch([{$match: {'ns.db': 'shop', 'ns.coll': 'orders'}}]), resume_after=token. Roughly 30 lines and every one is documented.
• Senior signal. The interviewer wants to hear that you understand why the abstraction won — server-side filter + resume tokens + documented schema — not just that you use it.

Question. A team maintains a legacy Mongo 3.4 oplog tailer that reads shop.orders and publishes inserts, updates, and deletes to a downstream queue. They upgrade to Mongo 6.0. Rewrite the ingest as a native change-stream consumer in Python, and quantify what code disappears.

Input.

Component	Legacy (3.4 oplog tailer)	Modern (change stream)
Cursor open	TAILABLE_AWAIT on local.oplog.rs	collection.watch([...])
Filter	client-side if op['ns'] == 'shop.orders'	server-side $match in pipeline
Delete hydration	side cache of _id → document	fullDocument: 'updateLookup' for updates; delete carries documentKey
Resume	per-shard Timestamp tracker	opaque resume_token
Sharded coordination	write your own merger	mongos handles it
Line count	~300	~30

Code.

# Legacy — oplog tailer (Mongo 3.4)
from pymongo import MongoClient, CursorType
from bson.timestamp import Timestamp

client = MongoClient("mongodb://legacy-primary:27017/?replicaSet=rs0")
oplog  = client.local.oplog.rs

last_ts = load_last_ts_from_disk()   # per-shard Timestamp checkpoint

cursor = oplog.find(
    {"ts": {"$gt": last_ts}, "ns": "shop.orders"},
    cursor_type=CursorType.TAILABLE_AWAIT,
    oplog_replay=True,
).max_await_time_ms(1000)

for entry in cursor:
    op = entry["op"]                 # 'i' | 'u' | 'd' | 'c' | 'n'
    if op == "i":
        publish_insert(entry["o"])
    elif op == "u":
        publish_update(entry["o2"]["_id"], entry["o"])   # o2 = query, o = mutation
    elif op == "d":
        deleted_doc = side_cache_lookup(entry["o"]["_id"])
        publish_delete(deleted_doc)
    save_last_ts_to_disk(entry["ts"])

# Modern — change-stream cursor (Mongo 6.0)
from pymongo import MongoClient

client = MongoClient("mongodb://modern-primary:27017/?replicaSet=rs0")
coll   = client.shop.orders

resume_token = load_resume_token_from_disk()

with coll.watch(
    pipeline=[],                              # empty = all ops on this coll
    full_document="updateLookup",             # hydrate updates
    resume_after=resume_token,                # None on first run
) as stream:
    for change in stream:
        op_type = change["operationType"]      # 'insert' | 'update' | 'replace' | 'delete' | ...
        if op_type == "insert":
            publish_insert(change["fullDocument"])
        elif op_type in ("update", "replace"):
            publish_update(change["documentKey"]["_id"], change["fullDocument"])
        elif op_type == "delete":
            publish_delete(change["documentKey"]["_id"])
        save_resume_token_to_disk(change["_id"])

Step-by-step explanation.

1. The legacy code opens local.oplog.rs with a tailable-await cursor and a Timestamp filter — that filter is a client-side range scan and the driver has to know Mongo-internal oplog fields (op, o, o2, ns, ts). Every one of those field names is undocumented API surface that broke in minor upgrades.
2. The delete branch is the ugliest: the oplog d operation only carries _id, so the pipeline needs a side cache populated at insert time to reconstruct the full deleted document. Missing the cache means missing routing information downstream.
3. The modern code opens collection.watch([]) with an empty aggregation pipeline (matches all ops on the collection) and a resume_after token. The driver hides the tailable cursor mechanics entirely.
4. full_document="updateLookup" tells the server to include the current document in the event payload for update events. Delete events carry documentKey (which always includes _id at minimum), so no side cache is needed to route the delete.
5. Every event's _id is the resume token. Persisting it on disk (or in a Mongo checkpoint collection, or in a Kafka offsets topic) after each downstream ack makes the loop restart-safe. On startup, resume_after=resume_token re-enters the stream at the same position.

Output.

Behaviour	Legacy tailer	Modern change stream
Lines of code	~300	~30
Undocumented field access	yes (op, o, o2)	no (documented schema)
Delete hydration	side cache required	documentKey on the event
Sharded coordination	write your own merger	mongos handles it
Resume semantics	per-shard Timestamp tracker	opaque resume_token
Upgrade risk	breaks on every minor Mongo version	stable driver API

Rule of thumb. Never write a new oplog tailer. Any pipeline still tailing local.oplog.rs in 2026 is technical debt with a countdown timer — port to change streams before the next Mongo major release makes the legacy code silently wrong.

Worked example — the four axes on a whiteboard

Detailed explanation. A common senior interview opener: "Draw the four things you need to answer to build a Mongo CDC pipeline." The four axes are (1) event schema, (2) resume token, (3) full-document strategy, (4) sharded coordination. Show a candidate's whiteboard sketch answering each and how they interact — because interviewers grade the connections between axes, not the axes in isolation.

• Axis 1 — schema. What fields does a change event contain, and how do consumers route from them?
• Axis 2 — token. How do we persist position across restarts without losing or duplicating events?
• Axis 3 — full doc. Do consumers need the full document, and if so, what does it cost to fetch it?
• Axis 4 — sharding. Where does the watch live in a sharded cluster, and what is the ordering guarantee?

Question. A senior interviewer says "Sketch me the four axes of a Mongo change-stream ingest. Show the schema, where the token lives, how you choose the full-document mode, and where the watch runs against a sharded cluster."

Input.

Axis	Question the axis answers	Where the decision lives
Schema	"What fields does the consumer parse?"	Server (fixed by driver API)
Resume token	"Where does the consumer persist position?"	Consumer (checkpoint doc / Kafka offset)
Full-document mode	"Does the consumer need the current row?"	Consumer (fullDocument option)
Sharded coordination	"Where does watch() open?"	Deployment topology (mongos in sharded clusters)

Code.

+----------------------- Axis 1: Change Event Schema -----------------------+
|  {                                                                        |
|    _id: <resume token>,               # opaque cursor position            |
|    operationType: "insert" | "update" | "replace" | "delete" |            |
|                   "drop" | "rename" | "invalidate",                       |
|    ns: {db: "shop", coll: "orders"},                                      |
|    documentKey: {_id: ObjectId(...), shard_key: ...},                     |
|    fullDocument: {...},               # present if updateLookup or insert |
|    updateDescription: {                                                   |
|      updatedFields: {"status": "shipped"},                                |
|      removedFields: [],                                                   |
|      truncatedArrays: []                                                  |
|    },                                                                     |
|    clusterTime: Timestamp(1720339200, 5),                                 |
|    wallTime: ISODate("2026-07-07T12:00:00.005Z")                          |
|  }                                                                        |
+---------------------------------------------------------------------------+

+----------------------- Axis 2: Resume Token Loop -------------------------+
|   ┌──────────────┐    event      ┌────────────┐   ack      ┌──────────┐  |
|   │ change stream│ ─────────────▶ │ consumer   │ ─────────▶ │ sink     │  |
|   └──────▲───────┘                └─────┬──────┘            └──────────┘  |
|          │ resumeAfter                  │ persist token                    |
|          │                              ▼                                  |
|   ┌──────┴───────┐              ┌────────────┐                            |
|   │ oplog / shard│              │ checkpoint │                            |
|   └──────────────┘              └────────────┘                            |
+---------------------------------------------------------------------------+

+----------------------- Axis 3: Full-Document Mode ------------------------+
|                                                                           |
|  default              →  delta only (updateDescription)                   |
|  updateLookup         →  round-trip to primary, current state             |
|  required (6.0+)      →  error if pre/post image unavailable              |
|  whenAvailable (6.0+) →  best-effort using pre-image cache                |
+---------------------------------------------------------------------------+

+----------------------- Axis 4: Sharded Watch -----------------------------+
|                                                                           |
|      ┌────────┐                                                          |
|      │ mongos │◀──── watch() opens here in sharded clusters              |
|      └───┬────┘                                                          |
|          │ fan out                                                       |
|   ┌──────┼──────┐                                                        |
|   ▼      ▼      ▼                                                        |
| shard0 shard1 shard2   ← each shard streams its oplog slice              |
|   │      │      │                                                        |
|   └──────┴──────┘                                                        |
|          │                                                                |
|     merge by clusterTime → globally ordered event stream                 |
+---------------------------------------------------------------------------+

Step-by-step explanation.

1. The event schema is the anchor — every other decision references a field on the event. operationType drives routing, documentKey._id drives downstream keys, updateDescription drives delta application, clusterTime drives ordering, and _id drives the resume-token loop.
2. The resume-token loop is the invariant: event flows out to the consumer, the consumer routes it to a sink, the sink acks the write, the consumer then persists the token. Persisting before the ack duplicates on retry; persisting the wrong token drops events on restart.
3. The full-document decision is a per-consumer knob, not a global toggle. A warehouse sink usually wants updateLookup so every event carries the current row. A raw-event log wants the default delta and pays no round-trip. Some consumers want both — split them into two watch() cursors.
4. The sharded watch runs against mongos, which fans the watch out to every shard's oplog slice and merges by cluster time on the way back. The merge is what makes the stream globally ordered — but it costs a small amount of latency proportional to the slowest shard.
5. The four axes are interlocked: a warehouse sink (axis 3 → updateLookup) reading a sharded cluster (axis 4 → mongos) with restart safety (axis 2 → checkpoint) parsing the documented schema (axis 1) is the canonical shape of a production Mongo CDC pipeline in 2026. Every senior interview probes at least three of the four in the first ten minutes.

Output.

Axis	What good looks like	What "amateur" looks like
Schema	Name every field without prompting	Confuses documentKey with fullDocument
Token	Persist after ack; store per stream	Persist before ack; lose tokens on crash
Full doc	Choose per consumer, not globally	Enable updateLookup everywhere by default
Sharding	Watch on mongos; expect merged ordering	Watch each shard separately and re-merge in app

Rule of thumb. When an interviewer asks about Mongo CDC, structure the whole answer around the four axes. Even if the follow-ups get deep into any single axis, having the four-part scaffold in your head keeps the answer coherent and avoids the "I know pieces but can't sequence them" impression.

Worked example — sizing the oplog retention window

Detailed explanation. Resume tokens work only while the oplog still contains the position they point to. If your consumer falls behind and the oplog rolls over past the checkpoint, the next resumeAfter call raises ChangeStreamHistoryLost and the consumer has to re-snapshot the collection. Sizing the oplog is a first-order design decision that senior engineers own — get it wrong and you lose events during your worst outage.

• Default oplog size. WiredTiger default is min(50 GB, 5% of free disk) — often good enough for small deployments, disastrous for large write-heavy ones.
• Retention time vs size. What matters to a change-stream consumer is how many hours of history the oplog holds, not its byte size. High write throughput shortens retention proportionally.
• The formula. retention_hours = oplog_size_bytes / (write_bytes_per_hour). Measure the second value from db.getReplicationInfo() — Mongo reports it directly.

Question. A production replica set has a 100 GB oplog and averages 4 GB/hour of writes. The team wants at least 24 hours of retention so a consumer can crash for a full day without losing history. Do the math, verify with db.getReplicationInfo(), and describe what to do if the number is too small.

Input.

Parameter	Value
Oplog size (bytes)	100 GB
Write rate (bytes/hour)	4 GB/hour
Target retention	24 hours
Number of shards	1 (single replica set)

Code.

// Measure current oplog window
db.getReplicationInfo();
// {
//   "logSizeMB": 102400,
//   "usedMB": 78200,
//   "timeDiff": 87200,                 // seconds of history
//   "timeDiffHours": 24.22,
//   "tFirst": "Sat Jul 06 2026 12:00:00 GMT+0000",
//   "tLast":  "Sun Jul 07 2026 12:13:20 GMT+0000",
//   "now":    "Sun Jul 07 2026 12:13:20 GMT+0000"
// }

// If retention is too small, grow the oplog online (Mongo 3.6+ supports resize without restart)
db.adminCommand({ replSetResizeOplog: 1, size: 204800 });   // 200 GB target

// After resize, re-check
db.getReplicationInfo();

# Consumer-side guard rail — refuse to start if retention is too tight
from pymongo import MongoClient

client = MongoClient("mongodb://primary:27017/?replicaSet=rs0")
info   = client.admin.command("replSetGetStatus")
repl   = client.local.command("replicationInfo")   # or driver-specific helper

# Pseudo: min_retention_hours guard
MIN_RETENTION_HOURS = 24
observed = repl["timeDiffHours"]

if observed < MIN_RETENTION_HOURS:
    raise RuntimeError(
        f"Oplog retention {observed:.1f}h is below the required {MIN_RETENTION_HOURS}h. "
        "Resize the oplog before starting change-stream consumers."
    )

Step-by-step explanation.

1. db.getReplicationInfo() reports the actual oplog window in seconds and hours. The returned timeDiff = 87200 seconds = 24.22 hours means the oplog currently holds just over 24 hours of history under the current write rate. That is the number you compare against your consumer-outage SLO.
2. The formula check: 100 GB oplog / 4 GB per hour = 25 hours, which matches the reported 24.22 hours (some overhead reduces the theoretical maximum). Consistent numbers mean your write-rate measurement is sound.
3. If retention were only 6 hours, the mitigation ladder starts with (a) replSetResizeOplog to grow the oplog on the fly (Mongo 3.6+), (b) reducing write rate at the source (large bulk imports pushed to off-hours), and (c) as a last resort provisioning larger disks.
4. The consumer-side guard rail refuses to start unless retention meets the target. This catches misconfiguration at deploy time rather than at outage time — the failure mode you want.
5. In sharded clusters, do this check per shard — each shard has its own oplog. A skewed shard with hot writes can have much shorter retention than the rest of the cluster. The consumer's effective window is the minimum across shards.

Output.

Measurement	Value	Meets 24h SLO?
Oplog size	100 GB	—
Write rate	4 GB/hour	—
Formula retention	25 hours	yes
db.getReplicationInfo()	24.22 hours	yes
After 2x write growth (8 GB/h)	12.5 hours	no — resize needed

Rule of thumb. Size the oplog for at least 24 hours of retention at your peak write rate, and monitor timeDiffHours in your dashboards. A shrinking oplog window is a leading indicator that your consumers will get ChangeStreamHistoryLost on the next outage.

Senior interview question on why change streams became the default

A senior interviewer might open with: "You inherit a legacy Mongo 3.4 CDC pipeline that tails local.oplog.rs directly, has intermittent duplicate deletes, and can't watch a sharded cluster. Walk me through the modern rewrite using change streams — the four axes you'd cover, the code shape, and the operational guardrails you'd add."

Solution Using a change-stream ingest wired to the four axes with a checkpoint-on-ack loop

# ingest.py — production Mongo CDC consumer using change streams
from pymongo import MongoClient
from pymongo.errors import PyMongoError, OperationFailure
import json, time, signal, logging

log = logging.getLogger("mongo.cdc")

CHECKPOINT_DB   = "cdc_ops"
CHECKPOINT_COLL = "resume_tokens"
STREAM_ID       = "shop.orders.v1"

def load_token(client):
    doc = client[CHECKPOINT_DB][CHECKPOINT_COLL].find_one({"_id": STREAM_ID})
    return doc["token"] if doc else None

def save_token(client, token):
    client[CHECKPOINT_DB][CHECKPOINT_COLL].update_one(
        {"_id": STREAM_ID},
        {"$set": {"token": token, "updated_at": time.time()}},
        upsert=True,
    )

def publish(event):
    # POST to sink, block until 2xx, raise on error → drives at-least-once
    ...

def run():
    client = MongoClient(
        "mongodb://mongos.internal:27017/?replicaSet=rs0",
        appname="orders-cdc-v1",
    )
    coll = client.shop.orders
    token = load_token(client)
    log.info("resuming after token=%s", "None" if token is None else "<opaque>")

    stopped = False
    signal.signal(signal.SIGTERM, lambda *_: setattr(run, "stopped", True))

    while not getattr(run, "stopped", False):
        try:
            with coll.watch(
                pipeline=[{"$match": {"operationType": {"$in":
                    ["insert", "update", "replace", "delete"]}}}],
                full_document="updateLookup",
                resume_after=token,
                max_await_time_ms=1000,
            ) as stream:
                for change in stream:
                    publish(change)                 # blocks until sink acks
                    token = change["_id"]
                    save_token(client, token)       # persist ONLY after ack
        except OperationFailure as e:
            if e.code == 286:   # ChangeStreamHistoryLost
                log.error("resume token stale; falling back to snapshot + fresh watch")
                snapshot_and_switch_to_current_time(client, coll)
                token = None
            else:
                log.exception("mongo error; will retry")
                time.sleep(2)
        except PyMongoError:
            log.exception("driver error; will retry")
            time.sleep(2)

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run()

Step-by-step trace.

Step	Axis	What the code does	Why it matters
Open watch on mongos	4 (sharded)	MongoClient(mongos.internal) + coll.watch(...)	Fans across shards; globally ordered
Server-side $match	1 (schema)	Filter to CRUD ops in pipeline	Skip drop/rename noise
full_document=updateLookup	3 (full doc)	Hydrate updates in the event	Sink sees current row without extra call
resume_after=token	2 (token)	Reopen at last committed position	Restart-safe
Persist token AFTER publish	2 (token)	Save order matters	At-least-once, no drop
Handle ChangeStreamHistoryLost	2 (token)	Fall back to snapshot	Graceful degradation

After deployment, the consumer runs indefinitely, checkpoints per event, and restarts cleanly on SIGTERM. The 30-line loop replaces the ~300-line oplog tailer. Duplicate deletes disappear because the documentKey on the event carries _id directly, so no side cache is needed. Sharded coordination is free because mongos merges by cluster time on the server side.

Output:

Metric	Legacy tailer	Change-stream consumer
Lines of code	~300	~40
Duplicate-delete rate	non-zero	0
Sharded support	none	native
Restart safety	Timestamp bookkeeping	resume token
Undocumented field access	yes	no
ChangeStreamHistoryLost handling	crashed loop	snapshot fallback

Why this works — concept by concept:

• Watch on mongos — opens one logical cursor across the entire deployment. Ordering is guaranteed by cluster-time merge; the consumer sees a single monotonic stream regardless of shard count.
• Server-side $match — pushes filtering to the server. Bytes over the wire are the interesting events only. In a busy collection with rare interesting updates, this alone cuts network cost by 10–100x.
• Full-document lookup for warehouse sinks — updateLookup returns the current document from the primary as of the read, hydrating an update event with the row a warehouse actually wants. Delete events don't need it because documentKey already carries _id.
• Checkpoint on ack — persisting the resume token after the sink acks the write is what makes the loop at-least-once. If the process dies between publish and save, the next start reprocesses the last event — better than dropping it.
• ChangeStreamHistoryLost fallback — the graceful-degradation branch snapshots the collection to bootstrap the sink to a consistent point, then reopens the stream with startAtOperationTime = now. Skipping this branch means outages > oplog retention permanently break the pipeline.
• Cost — one long-lived cursor per stream + one round-trip per update event under updateLookup. CPU on the consumer is dominated by JSON encoding to the sink; Mongo-side cost is dominated by the updateLookup reads, which land on the primary and count against its working-set memory. Big-O is O(events) at runtime; O(1) at startup for token load.

Streaming
Topic — streaming
Streaming ingest and change-data-capture problems

Practice →

ETL Topic — etl ETL problems on document-store CDC pipelines

Practice →

---

2. Change event schema + operationType

Every change event is a documented envelope — parse the six required fields, then branch on operationType

The mental model in one line: every change-stream event is a BSON envelope with a fixed set of top-level fields — _id, operationType, ns, documentKey, clusterTime, and one or more type-specific payloads (fullDocument, updateDescription, to, truncatedArrays) — and the consumer's job is to branch on operationType and use the payload fields that apply to that branch. Knowing every field cold is what separates a senior CDC engineer from a bootcamp graduate reading the docs during the interview.

The six required top-level fields.

• _id. The resume token. An opaque BSON document you never look inside; you only store it and pass it back to resumeAfter. The exact shape ({"_data": "826..."}) is an implementation detail.
• operationType. A string that names the kind of change. Values in Mongo 6.0: insert, update, replace, delete, drop, dropDatabase, rename, invalidate. Newer versions add create, createIndexes, modify, shardCollection, refineCollectionShardKey, reshardCollection.
• ns. A subdocument {db, coll} naming the namespace. For dropDatabase and invalidate, only db is present. For cluster-wide watches this is how you route across collections.
• documentKey. For document-affecting operations (insert, update, replace, delete), the _id of the affected document — plus every shard-key field if the collection is sharded. {"_id": ObjectId(...), "tenant_id": 42} for a shard-key-including collection.
• clusterTime. A Timestamp(seconds, ordinal) that positions the event in cluster time. Used for ordering, for startAtOperationTime, and for downstream event-time processing.
• wallTime (6.0+). An ISODate estimate of when the operation happened. Approximate; use for logging, never for ordering.

The type-specific payload fields.

• fullDocument. For insert and replace, always present (the full new document). For update, present only if fullDocument: 'updateLookup' or 'required' is set. For delete, present only if fullDocumentBeforeChange is enabled and a pre-image exists.
• updateDescription. For update only. Subdocument with updatedFields (map of field paths to new values), removedFields (array of unset field paths), truncatedArrays (array of {field, newSize} for array truncations, 5.0+), and disambiguatedPaths (6.1+, for dotted field paths).
• to. For rename only. A {db, coll} naming the new namespace.
• documentKey. For delete, this is how the consumer knows which row was deleted — there is no fullDocument unless a pre-image was captured.
• fullDocumentBeforeChange (6.0+). The pre-image of an updated/deleted document, if pre-image capture is enabled on the collection.

The operationType values a consumer must handle.

• insert. New document written. fullDocument is always present. Route to sink as a new row.
• update. Partial modification to an existing document. updateDescription describes the delta; fullDocument is present only under updateLookup. Route as an update or an upsert.
• replace. Whole-document replacement (db.coll.replaceOne(...)). fullDocument is always present; there is no updateDescription. Semantically equivalent to a delete+insert of the same _id.
• delete. Document removed. Only documentKey (and fullDocumentBeforeChange if enabled). Route as a tombstone.
• drop. The watched collection was dropped. Followed immediately by an invalidate on collection-scoped watches; on database or cluster-scoped watches the stream continues.
• rename. Collection renamed. ns is the old namespace, to is the new one.
• invalidate. The stream is dead — the underlying resource was dropped or the watch is otherwise unusable. Consumer must open a fresh stream (usually against startAtOperationTime).
• dropDatabase. A database was dropped. Followed by invalidate on any watches over that database.

The ns.db, ns.coll, and cluster-wide routing story.

• Collection-scoped watch. db.coll.watch(...). ns is always {db: 'shop', coll: 'orders'} on every event.
• Database-scoped watch. db.watch(...). ns varies by collection; consumer routes on ns.coll.
• Cluster-wide watch. client.watch(...). ns varies by (db, coll); consumer routes on both.
• Server-side pipeline. $match, $project, $replaceRoot, $redact, $addFields, $set, $unset, $replaceWith in the pipeline all filter/shape events server-side. Use [{$match: {'ns.coll': 'orders', 'operationType': 'insert'}}] to skip everything else.

Common interview probes on schema.

• "Name the top-level fields on a change event." — insert, update, replace, delete, drop, rename, invalidate is the answer they want, plus _id, operationType, ns, documentKey, clusterTime.
• "What's on a delete event that lets you route it downstream?" — documentKey (and pre-image if enabled).
• "How does an update event differ from a replace event?" — update has updateDescription (delta); replace has fullDocument (whole new doc); replace has no delta.
• "What's the difference between drop and invalidate?" — drop is the operation; invalidate is the "your stream is dead" signal on watches scoped to the dropped resource.

Worked example — parse a batch of five change events

Detailed explanation. A concrete parsing exercise: given a batch of five real change events pulled from a stream, extract the operation type, the target _id, and the routable payload for each. This is the kind of drill an interviewer runs to check that a candidate can read the schema off a page, not just recite it.

• Events span the four core ops. insert, update (delta only), update (with updateLookup), delete, replace.
• Routing target for each. Insert → new row; update → delta apply; replace → upsert; delete → tombstone.
• Extra credit. Notice that documentKey on the delete carries the shard key too.

Question. Given the five events below, write a Python router that dispatches to handle_insert, handle_update, handle_replace, and handle_delete. For updates, log whether the event carries a full document.

Input.

#	operationType	Notes
1	insert	new order for tenant 42
2	update	delta only — status changed
3	update	with fullDocument (updateLookup)
4	delete	delete of an existing order
5	replace	wholesale replace of the doc

Code.

{"_id": {"_data": "826..."},
 "operationType": "insert",
 "ns": {"db": "shop", "coll": "orders"},
 "documentKey": {"_id": {"$oid": "66a1..."}, "tenant_id": 42},
 "fullDocument": {"_id": {"$oid": "66a1..."}, "tenant_id": 42,
                  "status": "pending", "total": 129.99, "created_at": "2026-07-07T12:00:00Z"},
 "clusterTime": {"$timestamp": {"t": 1720339200, "i": 1}},
 "wallTime": "2026-07-07T12:00:00.005Z"}

{"_id": {"_data": "827..."},
 "operationType": "update",
 "ns": {"db": "shop", "coll": "orders"},
 "documentKey": {"_id": {"$oid": "66a1..."}, "tenant_id": 42},
 "updateDescription": {"updatedFields": {"status": "shipped"},
                       "removedFields": [], "truncatedArrays": []},
 "clusterTime": {"$timestamp": {"t": 1720339260, "i": 3}},
 "wallTime": "2026-07-07T12:01:00.011Z"}

{"_id": {"_data": "828..."},
 "operationType": "update",
 "ns": {"db": "shop", "coll": "orders"},
 "documentKey": {"_id": {"$oid": "66a2..."}, "tenant_id": 42},
 "fullDocument": {"_id": {"$oid": "66a2..."}, "tenant_id": 42,
                  "status": "delivered", "total": 45.00,
                  "created_at": "2026-07-06T09:15:00Z"},
 "updateDescription": {"updatedFields": {"status": "delivered"},
                       "removedFields": [], "truncatedArrays": []},
 "clusterTime": {"$timestamp": {"t": 1720339320, "i": 2}},
 "wallTime": "2026-07-07T12:02:00.021Z"}

{"_id": {"_data": "829..."},
 "operationType": "delete",
 "ns": {"db": "shop", "coll": "orders"},
 "documentKey": {"_id": {"$oid": "66a3..."}, "tenant_id": 42},
 "clusterTime": {"$timestamp": {"t": 1720339380, "i": 1}},
 "wallTime": "2026-07-07T12:03:00.007Z"}

{"_id": {"_data": "82a..."},
 "operationType": "replace",
 "ns": {"db": "shop", "coll": "orders"},
 "documentKey": {"_id": {"$oid": "66a4..."}, "tenant_id": 42},
 "fullDocument": {"_id": {"$oid": "66a4..."}, "tenant_id": 42,
                  "status": "returned", "total": 0.00, "created_at": "2026-07-05T10:00:00Z"},
 "clusterTime": {"$timestamp": {"t": 1720339440, "i": 4}},
 "wallTime": "2026-07-07T12:04:00.033Z"}

# router.py — dispatch on operationType
import logging
log = logging.getLogger("cdc.router")

def route(change):
    op = change["operationType"]
    doc_id = change["documentKey"]["_id"]
    if op == "insert":
        handle_insert(doc_id, change["fullDocument"])
    elif op == "update":
        has_full = "fullDocument" in change
        log.info("update _id=%s hasFullDoc=%s delta=%s",
                 doc_id, has_full, change["updateDescription"]["updatedFields"])
        handle_update(doc_id, change["updateDescription"],
                      change.get("fullDocument"))
    elif op == "replace":
        handle_replace(doc_id, change["fullDocument"])
    elif op == "delete":
        handle_delete(doc_id, change["documentKey"])   # shard key travels along
    else:
        log.warning("unhandled operationType=%s", op)

Step-by-step explanation.

1. Event 1 is an insert. fullDocument is always present on inserts — the sink can write the row directly with no lookup. The consumer extracts documentKey._id for the primary key and passes the full document body.
2. Event 2 is a delta-only update. updateDescription.updatedFields says {status: "shipped"} was set. There is no fullDocument because updateLookup was not enabled for this event. Downstream must merge this delta into whatever state it holds.
3. Event 3 is an update with updateLookup enabled — both updateDescription and fullDocument are present. The downstream sink can either replay the delta or use the full document; the correctness story is the same, but the full-document path is cheaper if the sink is an upsert.
4. Event 4 is a delete. Only documentKey is present. Note that documentKey includes tenant_id — the shard key — which lets the router send the tombstone to the correct downstream partition without a lookup.
5. Event 5 is a replace. fullDocument is the whole new document; there is no updateDescription. Semantically a delete+insert of the same _id. Sinks that upsert see no difference from an insert; sinks that maintain a change log see one event rather than two.

Output.

#	op	routed as	payload used	notes
1	insert	new row	fullDocument	direct write
2	update	delta apply	updateDescription	no fullDocument
3	update	upsert	fullDocument (preferred)	has both
4	delete	tombstone	documentKey	shard key travels
5	replace	upsert	fullDocument	no delta needed

Rule of thumb. Build the router as a single operationType switch and let each branch pull only the fields it needs. Do not try to unify insert/update/replace into a single "upsert" path early — the schema differences matter for correctness in warehouse and search sinks.

Worked example — server-side aggregation pipeline filtering

Detailed explanation. A busy collection emits thousands of events per second, but a downstream consumer only cares about updates that touch status or total. Pushing that filter into the server-side aggregation pipeline saves 90%+ of network traffic and 90%+ of consumer CPU. This is the single highest-leverage optimisation on the change-stream API.

• The pipeline. Passed as the first argument to watch(). Any read-only aggregation stages: $match, $project, $replaceRoot, $addFields, $redact, $unset, $set, $replaceWith.
• What breaks. Write stages ($out, $merge) are forbidden. Stages that reorder events break ordering — avoid $sort and $group.
• The win. Server does the filter; only matching events cross the wire.

Question. Write a $match stage that keeps only insert events and update events where the update touches status or total, then project only the fields the consumer needs.

Input.

Requirement	Value
Keep operationTypes	insert, update
Update filter	updatedFields.status OR updatedFields.total present
Projection	keep _id, operationType, ns, documentKey, fullDocument, updateDescription

Code.

from pymongo import MongoClient

client = MongoClient("mongodb://mongos.internal:27017/")
coll   = client.shop.orders

pipeline = [
    {"$match": {
        "$or": [
            {"operationType": "insert"},
            {"operationType": "update",
             "$or": [
                {"updateDescription.updatedFields.status": {"$exists": True}},
                {"updateDescription.updatedFields.total":  {"$exists": True}},
             ]},
        ],
    }},
    {"$project": {
        "_id": 1,
        "operationType": 1,
        "ns": 1,
        "documentKey": 1,
        "fullDocument": 1,
        "updateDescription": 1,
        "clusterTime": 1,
    }},
]

with coll.watch(pipeline=pipeline, full_document="updateLookup") as stream:
    for change in stream:
        route(change)   # from the previous worked example

// mongosh equivalent — same pipeline shape
db.orders.watch([
  { $match: { $or: [
      { operationType: "insert" },
      { operationType: "update",
        $or: [
          { "updateDescription.updatedFields.status": { $exists: true } },
          { "updateDescription.updatedFields.total":  { $exists: true } },
        ] },
  ] } },
  { $project: {
      _id: 1, operationType: 1, ns: 1, documentKey: 1,
      fullDocument: 1, updateDescription: 1, clusterTime: 1,
  } },
], { fullDocument: "updateLookup" });

Step-by-step explanation.

1. The $match stage uses $or to keep two shapes: pure inserts (any insert is interesting), and updates whose updatedFields contains status or total. Field-path syntax in the match uses dotted paths against the change event's own fields — server evaluates this before shipping bytes.
2. The nested $or on the update branch uses $exists: true to test for the presence of the field in updatedFields. Values don't matter; the fact that the update touched the field is enough to keep it.
3. $project is a small optimisation. Every field kept in the projection is a field that crosses the wire. Dropping wallTime (approximate, only useful for logging) and any custom fields the driver might add saves bytes at high event rates.
4. The full_document="updateLookup" option is applied at the watch() call, not in the pipeline. It runs after the pipeline — Mongo evaluates $match first, decides the event is interesting, then does the updateLookup round-trip, then applies the projection.
5. The net effect: the change stream now emits maybe 1% of the physical event volume, and each event that does emit has already been trimmed to the fields the router uses. Consumer CPU and network cost drop by orders of magnitude, and the updateLookup cost applies only to events that survive the filter — a huge secondary win.

Output.

Stage	Reduces	By how much (typical)
$match on operationType	wire bytes	~50-70% (drop drop/rename/invalidate)
$match on updatedFields	wire bytes	~90-99% for narrow trigger fields
$project slim	wire bytes	~10-30% (drop unused fields)
updateLookup on filtered events	primary reads	proportional to filter selectivity

Rule of thumb. Any consumer that cares about a specific field on updates should push a $match into the pipeline. Every wire byte saved is a byte your consumer doesn't parse and a byte Mongo doesn't serialize.

Worked example — routing a cluster-wide watch by ns

Detailed explanation. A single ingest service watches an entire Mongo deployment (three databases, twelve collections) and routes each event to the correct downstream topic. The router dispatches on ns.db and ns.coll; the same event schema applies uniformly whether the watch is collection, database, or cluster scope.

• Watch scope. client.watch(...) — the highest scope. Sees every event across every db and collection except system collections.
• Routing. By ns.db and ns.coll into a dispatch table.
• Topic naming. mongo.<db>.<coll> is a common convention, matching Debezium's naming.

Question. Build a router that takes any change event from a cluster-wide watch and dispatches to a Kafka topic named mongo.<db>.<coll>, dropping events for the admin, local, and config databases.

Input.

Database	Collections	Route?
shop	orders, customers, products	yes
billing	invoices, payments	yes
analytics	events, sessions	yes
admin, local, config	*	drop (system)

Code.

from pymongo import MongoClient
from kafka import KafkaProducer
import json

SYSTEM_DBS = {"admin", "local", "config"}

client = MongoClient("mongodb://mongos.internal:27017/")
producer = KafkaProducer(
    bootstrap_servers="kafka.internal:9092",
    value_serializer=lambda v: json.dumps(v, default=str).encode(),
    key_serializer=lambda k: k.encode() if isinstance(k, str) else k,
    linger_ms=10,
    acks="all",
)

pipeline = [{"$match": {"ns.db": {"$nin": list(SYSTEM_DBS)}}}]

with client.watch(pipeline=pipeline, full_document="updateLookup") as stream:
    for change in stream:
        db_name   = change["ns"]["db"]
        coll_name = change["ns"].get("coll")
        if not coll_name:
            continue                          # dropDatabase, invalidate
        topic = f"mongo.{db_name}.{coll_name}"
        key   = str(change["documentKey"]["_id"])
        producer.send(topic, key=key, value=change)
        producer.flush()                      # per-event flush for at-least-once
        # persist token AFTER flush — omitted for brevity

# Kafka topic layout — one topic per (db, coll)
mongo.shop.orders
mongo.shop.customers
mongo.shop.products
mongo.billing.invoices
mongo.billing.payments
mongo.analytics.events
mongo.analytics.sessions

Step-by-step explanation.

1. client.watch(...) opens a cluster-wide watch against mongos. Every event on every non-system collection flows through this one cursor. This is the simplest possible ingest topology — one process, one cursor, N topics.
2. The server-side $match on ns.db not in system_dbs drops the 99% of events (mostly admin.system.* heartbeat noise) before they cross the wire. This is a critical filter — without it, the consumer wastes CPU parsing events it will immediately drop.
3. The router pulls ns.db and ns.coll off every event and constructs the topic name. Events without a coll field (dropDatabase, invalidate, some cluster-scoped ops) are skipped.
4. The message key is the document _id (serialized to string). This ensures updates to the same document land on the same Kafka partition, preserving per-document ordering on the sink side — essential for downstream consumers doing state derivation.
5. producer.flush() per event is the naive at-least-once path — every event acks before the token is persisted. In production, batch flushes plus token persistence in the same "commit" batch drops the round-trip cost 10-100x. The naive shape here is the interview-safe answer; the batched shape is a follow-up conversation.

Output.

Metric	Value
Cursors open	1 (cluster-wide watch)
Topics fed	N (one per non-system collection)
Per-event Kafka round-trips	1 (naive) or 1 per batch (optimised)
Per-document ordering preserved	yes (key = _id)
System-db events dropped server-side	yes

Rule of thumb. Use client.watch(...) when the number of watched collections is small enough that one cursor's throughput isn't a bottleneck (typically <50 collections, <5k events/sec). Beyond that, shard the ingest by database or collection with separate watch() cursors and separate resume tokens.

Senior interview question on change event schema

A senior interviewer might ask: "Draw the change event schema on the whiteboard. Then show me the code that routes an event to a Kafka topic, including the server-side filter that keeps only updates touching a specific set of fields."

Solution Using a fielded router with a server-side aggregation pipeline

# router_solution.py — production event router with server-side filter
from pymongo import MongoClient
from kafka import KafkaProducer
import json, logging

log = logging.getLogger("cdc.router")

TRIGGER_FIELDS = ["status", "total", "billing.plan"]
SYSTEM_DBS     = {"admin", "local", "config"}

def build_pipeline():
    match_update = {"operationType": "update",
                    "$or": [{f"updateDescription.updatedFields.{f}": {"$exists": True}}
                            for f in TRIGGER_FIELDS]}
    return [
        {"$match": {
            "ns.db": {"$nin": list(SYSTEM_DBS)},
            "$or": [
                {"operationType": {"$in": ["insert", "replace", "delete"]}},
                match_update,
            ],
        }},
        {"$project": {
            "_id": 1, "operationType": 1, "ns": 1, "documentKey": 1,
            "fullDocument": 1, "updateDescription": 1, "clusterTime": 1,
        }},
    ]

def dispatch(change, producer):
    db_name   = change["ns"]["db"]
    coll_name = change["ns"].get("coll")
    if not coll_name:
        return
    topic = f"mongo.{db_name}.{coll_name}"
    key   = str(change["documentKey"]["_id"])
    envelope = {
        "op":        change["operationType"],
        "id":        key,
        "before":    change.get("fullDocumentBeforeChange"),
        "after":     change.get("fullDocument"),
        "patch":     change.get("updateDescription"),
        "source":    {"db": db_name, "coll": coll_name,
                      "ts": str(change["clusterTime"])},
    }
    producer.send(topic, key=key, value=envelope)

def run():
    client = MongoClient("mongodb://mongos.internal:27017/", appname="router-v1")
    producer = KafkaProducer(
        bootstrap_servers="kafka.internal:9092",
        value_serializer=lambda v: json.dumps(v, default=str).encode(),
        key_serializer=lambda k: k.encode(),
        linger_ms=20, acks="all",
    )
    with client.watch(pipeline=build_pipeline(),
                      full_document="updateLookup") as stream:
        for change in stream:
            try:
                dispatch(change, producer)
                producer.flush()
                save_resume_token(change["_id"])
            except Exception:
                log.exception("dispatch failed; will retry from last token")
                raise

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    run()

Step-by-step trace.

Step	What happens	Field(s) consulted
Watch opens on mongos	Cluster-wide cursor	—
$match drops system dbs	Server-side filter	ns.db
$match keeps interesting ops	Insert/replace/delete always; update only if trigger fields touched	operationType, updateDescription.updatedFields.*
$project slims payload	Kept 7 fields	—
updateLookup hydrates updates	Only for events surviving filter	fullDocument
Router derives topic	mongo.<db>.<coll>	ns.db, ns.coll
Envelope carries op + id + patch	Debezium-style shape	all payload fields
Kafka key = documentKey._id	Preserves per-document ordering	documentKey._id
Token persisted after flush	At-least-once	_id

The router runs against a cluster-wide watch and pushes envelopes to per-collection Kafka topics. The server-side $match cuts wire volume by 90%+ in typical workloads; the updateLookup cost applies only to filtered events. The resume token is persisted after every flush, giving at-least-once semantics without dropping.

Output:

Metric	Value
Cursors open	1
Topics fed	N per (db, coll)
Wire-byte reduction vs no filter	~95%
updateLookup calls per second	rate × filter selectivity
Per-document ordering preserved	yes
At-least-once semantics	yes (token after ack)

Why this works — concept by concept:

• Server-side $match filter — pushes the operation-type and updated-field filter to the server. Wire bytes drop by 1-2 orders of magnitude; consumer parses only what it uses.
• Server-side $project — trims payload to the seven fields the router touches. Extra bytes are dropped in the server.
• Debezium-shape envelope — mirrors the widely known {op, id, before, after, patch, source} shape so downstream consumers can reuse existing Debezium tooling. before is only populated if pre-image capture is enabled.
• Kafka key = documentKey._id — hashes to a stable partition per document. All updates to a document land on the same partition, preserving per-document order end-to-end.
• Flush before token save — the flush blocks until the broker acknowledges the write. Persisting the resume token after the flush is what makes the pipeline at-least-once — a crash between flush and save reprocesses one event, but never drops.
• Cost — one cursor, one producer, one process. Consumer CPU is dominated by JSON encoding. Mongo-side cost is dominated by updateLookup reads for events that survive the filter. Big-O is O(matched_events) for network and CPU; O(filter_evaluations) for wire byte reduction.

Streaming
Topic — streaming
Streaming schema-parsing and event routing problems

Practice →

ETL Topic — etl ETL problems on filter-and-project pipelines

Practice →

---

3. Resume tokens + at-least-once semantics

The resume token is the single most-interviewed detail on change streams — miss it and your pipeline drops events on every restart

The mental model in one line: a resume token is an opaque BSON blob that encodes a position in the oplog (and cluster time in sharded deployments), the driver returns one on every change event, and the whole game is to persist that token after the downstream sink acknowledges the event so that a crashed consumer restarts exactly at the next un-processed event. Every senior CDC bug traces back to persisting the token at the wrong moment or losing the token altogether.

The three resume options — resumeAfter, startAfter, startAtOperationTime.

• resumeAfter. The classic. Pass the token of the last event you processed; the stream re-opens starting with the next event. Fails with ChangeStreamHistoryLost if the oplog no longer contains that position.
• startAfter. Newer variant (Mongo 4.2+). Semantically similar to resumeAfter, but can survive an invalidate event — if the underlying collection was dropped and re-created, startAfter continues on the new one, resumeAfter errors.
• startAtOperationTime. Start at a cluster-time Timestamp instead of a token. Used on first startup (there is no token yet) or after ChangeStreamHistoryLost when you want to resume "as of now" and re-snapshot the historical state.

At-least-once vs exactly-once vs at-most-once.

• At-most-once. Persist the token before the sink ack. A crash between persist and ack drops the event. Never do this unless dropping is safe.
• At-least-once. Persist the token after the sink ack. A crash between ack and persist replays one event. The correct default; downstream sinks must be idempotent.
• Exactly-once. Requires a transactional coupling between the sink write and the token persist. Achievable only if the sink and the token store share a transaction (e.g. both in the same Postgres). Rare in practice.

Where to store the token.

• In a Mongo checkpoint collection. A dedicated cdc_ops.resume_tokens collection with one doc per stream. Simple, colocates the token with the source of truth.
• In Kafka offsets. For Kafka Connect / Debezium consumers, the token is stored in the connector's offsets topic. Transparent to the operator.
• In an external KV store. Redis, DynamoDB, etcd — same idea, different storage. Choose based on your team's existing operational patterns.
• On local disk. Only for stateless dev/prototyping. Never in production — a lost disk equals a lost token.

The ChangeStreamHistoryLost failure and its recovery.

• The error. OperationFailure: ChangeStream $changeStream stage is only valid for the current operation time. History has been lost. Error code 286.
• The cause. The oplog no longer contains the position the token points to. Usually because the consumer was down longer than the oplog retention window.
• The recovery. Bootstrap the sink to a consistent snapshot of the collection (or accept that some events are lost), then open a fresh stream with startAtOperationTime = current cluster time. This is a graceful degradation, not a data-safe recovery — the events between the token's position and "now" are lost forever.
• Prevention. Monitor the oplog window (db.getReplicationInfo().timeDiffHours) against your worst-case consumer downtime. Alert when the ratio drops below 2x.

Startup token-loading logic.

• First run — no token. Use startAtOperationTime with the current cluster time. No historical events; the stream starts from "now."
• Normal restart — token exists. Use resumeAfter=token (or startAfter=token on 4.2+).
• After ChangeStreamHistoryLost. Bootstrap the sink from a snapshot, then use startAtOperationTime = post-snapshot cluster time.

Common interview probes on resume tokens.

• "What's a resume token?" — opaque BSON blob encoding an oplog position.
• "When do you persist it — before or after the downstream ack?" — after, for at-least-once. Before is at-most-once.
• "What happens if the oplog rolls over past your token?" — ChangeStreamHistoryLost; recover by snapshotting the collection and starting at current cluster time.
• "Difference between resumeAfter and startAfter?" — startAfter survives invalidate events; resumeAfter does not.

Worked example — a consumer that restarts safely from the last token

Detailed explanation. The textbook shape: a consumer loop that loads a token on startup, streams events, publishes each to a sink, and persists the token after the sink acks. Show what a crash between "publish" and "save" does (replays one event, no drop), and what a crash between "save" and next "read" does (nothing — the token is already durable). This is the loop every senior CDC engineer keeps in muscle memory.

• The loop. load → open → for each event: publish → wait for ack → save token → repeat.
• The crash points. Between publish and save = replay; between save and next event = clean. Never drops.
• The invariant. The token in storage always points to an event that has been fully acknowledged downstream.

Question. Implement the consumer loop in Python with a Mongo-backed checkpoint collection. Show what happens on a crash after publish but before save, and after save but before the next event.

Input.

Component	Value
Source	Mongo replica set, watch on shop.orders
Sink	HTTP endpoint (must return 2xx to ack)
Checkpoint store	cdc_ops.resume_tokens
Stream ID	shop.orders.v1
Failure semantics	at-least-once

Code.

# consumer_safe_restart.py
from pymongo import MongoClient
from pymongo.errors import OperationFailure
import requests, time, logging, os

log = logging.getLogger("cdc")

STREAM_ID = "shop.orders.v1"
SINK_URL  = os.environ["SINK_URL"]

def load_token(client):
    d = client.cdc_ops.resume_tokens.find_one({"_id": STREAM_ID})
    return d["token"] if d else None

def save_token(client, token):
    client.cdc_ops.resume_tokens.update_one(
        {"_id": STREAM_ID},
        {"$set": {"token": token, "saved_at": time.time()}},
        upsert=True,
    )

def publish(event):
    # Sink must be idempotent — at-least-once may re-deliver
    r = requests.post(SINK_URL, json=event, timeout=5)
    r.raise_for_status()

def stream(client, coll, token):
    with coll.watch(resume_after=token,
                    full_document="updateLookup",
                    max_await_time_ms=1000) as cursor:
        for change in cursor:
            publish(change)               # (1) publish — may throw on sink error
            save_token(client, change["_id"])   # (2) save — only after publish success

def run():
    client = MongoClient("mongodb://mongos.internal:27017/")
    coll   = client.shop.orders
    while True:
        token = load_token(client)
        try:
            stream(client, coll, token)
        except OperationFailure as e:
            if e.code == 286:            # ChangeStreamHistoryLost
                log.error("history lost — snapshot and restart at current time")
                bootstrap_snapshot_and_reset_token(client, coll)
            else:
                raise
        except Exception:
            log.exception("transient error — retrying loop")
            time.sleep(2)

Step-by-step explanation.

1. load_token(client) on startup either returns the token from the checkpoint collection or None. None means "first run" — the driver treats resume_after=None as "start from now," matching startAtOperationTime = current.
2. Every event is published before the token is saved. If publish(event) raises (bad sink response, timeout), the token is not advanced. On the next loop iteration load_token returns the old token, and the stream re-opens starting at the same event. The sink sees the event again — at-least-once.
3. If the process crashes between step (1) and step (2) — after the sink acked but before the token was saved — the same replay happens. The next start reprocesses one event. The idempotent sink handles the duplicate.
4. If the process crashes between step (2) and the next cursor iteration — after the token was saved but before the next event was pulled — nothing happens. The next start loads the newly-saved token and picks up at the next un-processed event. Clean.
5. The ChangeStreamHistoryLost branch (error code 286) is the graceful degradation. The consumer falls back to a bootstrap snapshot and resets the token. Data between the old token and the snapshot cluster time is lost, but the pipeline recovers instead of hanging.

Output.

Failure point	Behaviour	Data loss	Duplicate
Before publish	retry with same token	none	none
During publish (sink error)	retry with same token	none	possible (idempotent sink handles)
After publish, before save_token	replay 1 event	none	1 duplicate (idempotent)
After save_token, before next event	resume normally	none	none
Oplog rolled past token	history lost → snapshot	possible	none

Rule of thumb. Persist the token after the sink acks. Always. The only exceptions are (a) the sink is not idempotent and dropping is acceptable (rare), or (b) you have transactional coupling between sink and token (rarer still). The default is at-least-once, and idempotency at the sink layer is the pipeline architect's job.

Worked example — the difference between resumeAfter and startAfter

Detailed explanation. The subtle-but-important distinction between resumeAfter and startAfter. Both take a token and both re-open the stream at that position. The difference: startAfter survives an invalidate event, so a stream that fires after the source collection was dropped and re-created will still resume; resumeAfter errors out on the same scenario.

• The scenario. Collection shop.orders is dropped and re-created (e.g. during a schema migration). The consumer holds a token from before the drop.
• resumeAfter=token. Errors immediately with a resume-related failure — the token references an oplog position that is no longer valid for the current (re-created) collection.
• startAfter=token. Continues normally on the re-created collection. Emits an insert event for every re-populated document.

Question. Given a collection-scoped watch on shop.orders and a hypothetical drop-and-recreate operation, show two consumer variants — one using resumeAfter, one using startAfter — and describe how each behaves.

Input.

Scenario	Value
Watch scope	db.shop.orders.watch(...)
Event 1	insert for order A
Event 2	update for order A
Ops after event 2	db.orders.drop() then db.orders.insertOne(...)
Consumer holds	resume token from event 2

Code.

# Variant A — resumeAfter
with coll.watch(resume_after=token_after_event_2) as cursor:
    for change in cursor:
        route(change)
# Behaviour: on the drop, cursor yields an `invalidate` event, cursor closes.
# Next open with resumeAfter=token_after_invalidate → error.

# Variant B — startAfter
with coll.watch(start_after=token_after_event_2) as cursor:
    for change in cursor:
        route(change)
# Behaviour: on the drop, cursor yields `invalidate`, cursor closes.
# Next open with startAfter=token_after_invalidate → continues on new coll.
# Consumer sees new inserts on the re-created collection.

// mongosh — inspect resume behaviour
const cs = db.orders.watch([], { startAfter: lastToken });
while (cs.hasNext()) {
  const evt = cs.next();
  print(JSON.stringify(evt.operationType));
}

Step-by-step explanation.

1. Under resumeAfter, the consumer sees events 1 and 2 as expected. When the collection is dropped, an invalidate event fires and the cursor closes. The token from the invalidate event itself is now the last-processed token.
2. When the consumer restarts with resumeAfter=invalidate_token, the driver raises a resume-related error. The token references a resource (the old collection) that no longer exists in the oplog's addressable state. The consumer cannot proceed without manual intervention.
3. Under startAfter, the consumer sees the same events 1 and 2, the same invalidate, the same cursor close. But on the next open with startAfter=invalidate_token, the stream continues — starting on the newly-created collection with the same name.
4. This matters for schema-migration workflows where a collection is dropped and rebuilt. startAfter lets the consumer keep running through the migration without operator involvement. resumeAfter would require a code path that catches the error and forces a bootstrap.
5. In practice, use startAfter when your driver version supports it (Mongo 4.2+). It is a strict superset of resumeAfter — every scenario that works under resumeAfter also works under startAfter, plus the invalidate-survival case.

Output.

Scenario	resumeAfter behaviour	startAfter behaviour
Normal restart after processing events	resumes cleanly	resumes cleanly
After invalidate on dropped collection	errors on next open	continues on re-created collection
After ChangeStreamHistoryLost	errors	errors (both fail; must recover manually)
Driver version	3.6+	4.2+

Rule of thumb. Default to start_after (or startAfter) on Mongo 4.2+. It costs nothing on the happy path and gives you free recovery on the drop-and-recreate path. The only reason to keep resumeAfter in new code is compatibility with older driver versions.

Worked example — recovering from ChangeStreamHistoryLost with a snapshot

Detailed explanation. The consumer was down for 30 hours; the oplog retains only 24 hours. On restart, resumeAfter=old_token errors with ChangeStreamHistoryLost. The recovery ladder: (1) snapshot the collection at the current cluster time using a cursor with an explicit read concern, (2) push every snapshot document to the sink as a synthetic insert event, (3) open a fresh change stream with startAtOperationTime=snapshot_time, (4) resume streaming from there. Some events between old token and snapshot time are lost — that is the cost.

• The lost events. Everything between old_token_time and snapshot_time.
• The recovery output. Sink is consistent with the state of the collection at snapshot_time.
• The trade-off. Between "let the pipeline stay broken" and "lose some historical events." Always prefer recovery.

Question. Implement the recovery function. Show the snapshot, the synthetic insert events, and the change-stream reopen. Explain what happens if the collection is modified during the snapshot.

Input.

Parameter	Value
Old token	present, but invalid (history lost)
Oplog retention	24 h
Consumer downtime	30 h
Snapshot strategy	current cluster time, unindexed scan
Snapshot output	synthetic insert events to sink
Post-snapshot resume	startAtOperationTime = snapshot_time

Code.

from pymongo import MongoClient
from bson.timestamp import Timestamp
import time, logging

log = logging.getLogger("cdc.recovery")

def bootstrap_snapshot_and_reset_token(client, coll):
    # 1. Freeze snapshot time
    snapshot_time = current_cluster_time(client)   # Timestamp(sec, ord)
    log.warning("bootstrap snapshot at t=%s", snapshot_time)

    # 2. Read the collection at snapshot_time via an aggregation with atClusterTime
    #    (majority read concern gives us a consistent point-in-time view)
    with client.start_session(causal_consistency=False) as session:
        session.advance_cluster_time({"clusterTime": snapshot_time})
        cursor = coll.find({}, session=session)
        for doc in cursor:
            synthetic = {
                "operationType": "insert",
                "ns": {"db": coll.database.name, "coll": coll.name},
                "documentKey": {"_id": doc["_id"]},
                "fullDocument": doc,
                "clusterTime": snapshot_time,
                "_bootstrap": True,
            }
            publish(synthetic)                # sink upserts

    # 3. Reset the token — next open uses startAtOperationTime
    client.cdc_ops.resume_tokens.update_one(
        {"_id": "shop.orders.v1"},
        {"$set": {"token": None,
                  "start_at_operation_time": snapshot_time}},
        upsert=True,
    )

def current_cluster_time(client):
    hello = client.admin.command("hello")
    return hello.get("$clusterTime", {}).get("clusterTime")

def open_stream_after_bootstrap(coll, snapshot_time):
    with coll.watch(start_at_operation_time=snapshot_time,
                    full_document="updateLookup") as stream:
        for change in stream:
            publish(change)
            save_resume_token(change["_id"])

Step-by-step explanation.

1. Freeze the snapshot time as the current cluster time from the deployment. All subsequent reads target this timestamp so the sink sees a consistent point-in-time view.
2. Iterate over the collection with find({}) and emit a synthetic insert event for every document, marked _bootstrap: true so downstream can tell them apart from real inserts. The sink must be idempotent because the same _id may be re-inserted later during normal streaming if the document is modified.
3. During the snapshot, other writes can still happen — they aren't reflected in the snapshot (they happen after snapshot_time), but they will be replayed by the stream when we open with startAtOperationTime = snapshot_time. The net effect: sink converges to a consistent state without gaps beyond the initial "lost oplog history" window.
4. Reset the token in the checkpoint collection to indicate that the next open should use startAtOperationTime instead of resumeAfter. The token field is cleared; a new field records the snapshot cluster time.
5. Open the change stream with start_at_operation_time=snapshot_time. The stream begins at the snapshot's cluster time and replays every event since. Because the snapshot captured state at snapshot_time, the sink stays consistent — every post-snapshot mutation is streamed and applied on top.

Output.

Stage	Duration (5M doc coll)	Sink state
Snapshot scan	~30 min	~5M synthetic inserts applied
Post-snapshot stream open	<1 s	ready
Post-snapshot events replayed	~10 min (24 h of history compressed)	sink caught up
Total recovery time	~40 min	consistent

Rule of thumb. Every production Mongo CDC pipeline must ship the ChangeStreamHistoryLost recovery path from day one. Add it after your first outage and you'll wish you had it during that outage.

Senior interview question on resume tokens

A senior interviewer might ask: "Walk me through the resume-token lifecycle in a production Mongo CDC pipeline. Where do you store it, when do you save it, what happens when the oplog rolls past it, and how do you migrate from resumeAfter to startAfter?"

Solution Using a checkpoint-on-ack loop with graceful history-lost recovery

# ingest_full.py — production-shape ingest with token discipline
from pymongo import MongoClient
from pymongo.errors import OperationFailure
import logging, time, signal

log = logging.getLogger("cdc")
STREAM_ID = "shop.orders.v1"

class Checkpoint:
    def __init__(self, client):
        self.coll = client.cdc_ops.resume_tokens

    def load(self):
        d = self.coll.find_one({"_id": STREAM_ID}) or {}
        return d.get("token"), d.get("start_at_operation_time")

    def save(self, token):
        self.coll.update_one(
            {"_id": STREAM_ID},
            {"$set": {"token": token, "saved_at": time.time()},
             "$unset": {"start_at_operation_time": ""}},
            upsert=True,
        )

    def mark_bootstrap(self, snapshot_time):
        self.coll.update_one(
            {"_id": STREAM_ID},
            {"$set": {"token": None,
                      "start_at_operation_time": snapshot_time,
                      "bootstrap_at": time.time()}},
            upsert=True,
        )

def run(client, coll, sink):
    cp = Checkpoint(client)
    while True:
        token, start_at = cp.load()
        try:
            open_args = {"full_document": "updateLookup"}
            if token:
                open_args["start_after"] = token
            elif start_at:
                open_args["start_at_operation_time"] = start_at
            # first run: open with no resume — starts from "now"

            with coll.watch(**open_args) as cursor:
                for change in cursor:
                    sink.publish(change)         # blocks until acked
                    cp.save(change["_id"])       # persist AFTER ack
        except OperationFailure as e:
            if e.code == 286:  # ChangeStreamHistoryLost
                log.error("history lost — running bootstrap snapshot")
                snapshot_time = bootstrap_snapshot(client, coll, sink)
                cp.mark_bootstrap(snapshot_time)
            else:
                raise
        except Exception:
            log.exception("transient; retry")
            time.sleep(2)

Step-by-step trace.

Step	Value at start	After event 1	After event 2	After crash + restart
Token in Mongo	None	T1 (after publish 1)	T2 (after publish 2)	T2 (loaded)
Cursor position	opens at "now"	past event 1	past event 2	reopens at T2
Sink writes	0	1	2	2 (no dup on restart)
History-lost path	not triggered	not triggered	not triggered	not triggered
Bootstrap path	disabled	disabled	disabled	disabled

The loop is 30 lines of substantive code. Every event is published and then the token is saved — a crash between publish and save replays one event; the sink's idempotency handles the duplicate. startAfter is used when a token exists (survives invalidate); startAtOperationTime is used after a bootstrap. First run opens with neither, which the driver treats as "start from now" — no historical events are missed because the sink is bootstrapped in a separate operational step before the pipeline goes live.

Output:

Property	Value
Semantics	at-least-once
Idempotency required at sink	yes
History-lost recovery	bootstrap snapshot + startAtOperationTime
First-run behaviour	start from "now" (bootstrap sink separately)
Restart behaviour	resume from last saved token via startAfter
Token storage	Mongo checkpoint collection

Why this works — concept by concept:

• Token save after ack — the invariant that makes at-least-once work. Crash between publish and save = one replay; crash after save = clean. Never drops.
• startAfter over resumeAfter — free upgrade on Mongo 4.2+; survives collection drop/recreate during schema migrations.
• ChangeStreamHistoryLost branch — graceful degradation. Snapshot the collection, mark the checkpoint with startAtOperationTime = snapshot_time, reopen. Lost events between old token and snapshot are the cost of exceeding oplog retention.
• First-run "start from now" — the driver's default when no token or clusterTime is given. Pair with a separate operational step that bootstraps the sink to a consistent snapshot before the pipeline goes live.
• Checkpoint stored in Mongo — colocates the token with the source; no separate dependency to operate. Kafka offsets are equally valid for Kafka-Connect-hosted consumers.
• Cost — one extra Mongo write per event for the token save. Batched in high-throughput pipelines (save the token every N events with a bounded loss horizon). Big-O is O(events) at runtime; O(1) at startup for load; O(collection_size) once when history is lost and a bootstrap runs.

Streaming
Topic — streaming
Streaming checkpoint and at-least-once problems

Practice →

Optimization Topic — optimization Optimization problems on resume-token and offset tuning

Practice →

---

4. Full document lookup + updateDescription

updateDescription is what the oplog costs you; fullDocument: 'updateLookup' is what you pay to get the current row back

The mental model in one line: by default, a change-stream update event carries only the delta (updateDescription: {updatedFields, removedFields, truncatedArrays}) — the fields that actually changed — and the consumer must either apply that delta to its own state or pay a server round-trip via fullDocument: 'updateLookup' to receive the current document as of the read. The choice is a per-consumer trade-off between event size + latency (default) and sink simplicity + primary I/O (updateLookup).

The updateDescription payload — a delta representation.

• updatedFields. A subdocument mapping field paths to their new values. {"status": "shipped", "shipping.tracking_no": "1Z999"} after an $set: {status: "shipped", "shipping.tracking_no": "1Z999"} update.
• removedFields. An array of field paths that were removed via $unset. ["notes", "shipping.gift_wrap"].
• truncatedArrays (5.0+). An array of {field, newSize} entries for arrays whose length shrank. Only present when arrays are truncated (not when elements are modified in place).
• disambiguatedPaths (6.1+). Handles the ambiguity between object keys and array indices when field names are numeric.

The four fullDocument modes.

• default (omitted). No fullDocument on update events. updateDescription is present. Consumer applies deltas or does its own lookup if it wants the current state.
• 'updateLookup'. Server does a round-trip to the primary after the update; includes the current document in fullDocument. Not the state as of the update — the current state, which may already reflect later updates that haven't yet fired their own events. Practically consistent within a few ms.
• 'required' (6.0+). Errors if the pre/post image is not available for the change. Requires pre-image capture on the collection for fullDocumentBeforeChange; requires the document to still exist for fullDocument.
• 'whenAvailable' (6.0+). Best-effort: includes the pre/post image if available, silently omits if not. Non-erroring.

The cost model of updateLookup.

• One extra Mongo primary read per update event. Each update fires a findOne({_id}) behind the scenes.
• Not a point-in-time read. The lookup returns the current state of the document. If the document was updated three times in quick succession, all three events may return the same fullDocument (the state after the third update).
• Working-set impact. Every updateLookup reads through the primary's cache. High-cardinality updates on a large collection can evict genuine query-serving data.
• Latency impact. Adds one Mongo round-trip per event (~1-5 ms typical). Negligible for low-rate streams; significant at 10k+ events/sec.

When to enable updateLookup.

• Warehouse sink. Wants the current row; enable.
• Search index sink. Wants the current document; enable.
• Kafka fan-out to downstream consumers who may each want the full doc. Enable once at the source.
• Materialised view maintained by delta apply. Consumer applies updateDescription directly; do NOT enable.
• Raw event log for audit. Wants the delta as-is; do NOT enable.

Pre-image capture — fullDocumentBeforeChange.

• Enable per collection. db.runCommand({collMod: "orders", changeStreamPreAndPostImages: {enabled: true}}).
• Cost. Every update and delete copies the pre-image into an internal collection. Storage grows proportional to write rate + document size. Retention is bounded by preAndPostImagesCollectionOptions.expireAfterSeconds.
• Use case. Downstream that needs both before and after states (audit, CDC replay, GDPR delete-with-context).

Common interview probes on full document.

• "What's the default payload of an update event?" — updateDescription (delta), no fullDocument.
• "What does updateLookup cost?" — one extra primary read per event.
• "Is updateLookup point-in-time?" — no, it returns the current state, not the state at the update time.
• "When would you use whenAvailable over updateLookup?" — when you already run pre-image capture and want to be resilient to short-lived documents disappearing before the lookup.

Worked example — enrich delta events with fullDocument for a warehouse sink

Detailed explanation. A Snowflake-bound warehouse sink upserts the current row for every update event. Without updateLookup, the sink would have to fetch the row itself for every event — twice the round-trips because now both the CDC consumer and the warehouse loader are hitting Mongo. Enabling updateLookup at the source moves the round-trip to the change-stream server, which is closer to the oplog and cheaper than a driver-side re-read.

• Without updateLookup. Consumer receives delta, then the sink loader does a follow-up read.
• With updateLookup. Server hydrates the event; the sink loader upserts directly.
• The win. One round-trip instead of two, on the server side instead of the client side.

Question. Two ingest topologies. In topology A, updateLookup=None and the sink does its own read. In topology B, updateLookup='updateLookup' and the server hydrates. Compare event throughput and Mongo primary I/O for 1000 updates/sec on a 500-byte average document.

Input.

Metric	Topology A (no lookup)	Topology B (updateLookup)
Change events/sec	1000	1000
Extra reads from consumer	0	0
Extra reads from sink loader	1000	0
Extra reads from server (updateLookup)	0	1000
Event payload	delta only (~200 B)	delta + fullDocument (~700 B)
Wire bytes/sec	200 KB	700 KB

Code.

# Topology A — delta only, sink loader hydrates
from pymongo import MongoClient
from snowflake_loader import upsert_row

client = MongoClient("mongodb://mongos.internal:27017/")
coll   = client.shop.orders

with coll.watch(full_document=None) as stream:
    for change in stream:
        if change["operationType"] != "update":
            handle_non_update(change); continue
        doc_id = change["documentKey"]["_id"]
        # Sink loader does the read — 2 round trips per event
        current = coll.find_one({"_id": doc_id})
        upsert_row(current)

# Topology B — server hydrates via updateLookup
with coll.watch(full_document="updateLookup") as stream:
    for change in stream:
        if change["operationType"] != "update":
            handle_non_update(change); continue
        # fullDocument is on the event already — 1 round trip per event
        current = change["fullDocument"]
        upsert_row(current)

// Estimate the wire-byte cost per topology
const AVG_DOC = 500;   // bytes
const DELTA   = 200;   // bytes (event envelope + updateDescription)
const RATE    = 1000;  // events/sec

const bytesA = RATE * DELTA;              // = 200 KB/s  (change stream)
const readsA = RATE * AVG_DOC;            // = 500 KB/s  (sink hydrator)
const totalA = bytesA + readsA;           // = 700 KB/s

const bytesB = RATE * (DELTA + AVG_DOC);  // = 700 KB/s  (event carries doc)
const readsB = 0;                          // = 0        (no extra hydrator)
const totalB = bytesB + readsB;           // = 700 KB/s

// Same wire bytes! But B has ONE round-trip instead of TWO per event.

Step-by-step explanation.

1. Topology A costs two round-trips per event: one for the change-stream event, one for the sink loader's find_one. Wire bytes are split between the two channels. Latency is round-trip A + round-trip B.
2. Topology B costs one round-trip per event: the change-stream event already carries fullDocument, hydrated by the server. Total wire bytes are similar to A, but concentrated in one channel. Latency is round-trip B alone.
3. Mongo primary I/O is the same in both topologies — each update triggers one hydrating read regardless of who initiates it. The win of B is when the read happens (immediately after the update, on the server) versus when A does it (after the delta traverses the change-stream, on the consumer).
4. B also serialises the read at the server: if the same document is updated three times in one second, the server does three lookups but each one returns the current state, not the state at each of the three updates. Consumers using updateLookup for a warehouse sink don't care because upserting the current state three times converges to the same final state.
5. The trade-off flips when the sink is a materialised view maintained by delta apply. In that case, the delta is the payload; hydrating adds cost without benefit. Enable updateLookup per-consumer, not globally.

Output.

Property	Topology A	Topology B
Round-trips per event	2	1
Consumer code lines	~20	~10
Mongo primary reads/sec	1000 (from sink loader)	1000 (from server)
End-to-end p99 latency	~10 ms	~5 ms
Wire bytes/sec	~700 KB	~700 KB

Rule of thumb. For any sink that wants the current row, enable updateLookup at the source. It moves the read from the client side (extra network hop) to the server side (one hop). If the sink applies deltas instead, leave it off.

Worked example — delta apply for a materialised view

Detailed explanation. A materialised view of orders is maintained in Redis as a hash per order, keyed by _id. The consumer reads updateDescription, applies the delta to the Redis hash, and moves on. No fullDocument needed. This is the case where enabling updateLookup would add cost without benefit — the consumer never uses the hydrated document.

• The view. HSET order:{_id} status "shipped" total 129.99 ... — one Redis hash per order.
• The delta. updatedFields maps directly to Redis HSET field/value pairs; removedFields maps to HDEL.
• The efficiency. No extra Mongo reads. The updateDescription payload is the only payload the consumer touches.

Question. Implement the delta-apply consumer against Redis. Handle all three delta components (updatedFields, removedFields, truncatedArrays). Explain what happens when a document is updated by a nested-path operator like $set: {"shipping.tracking_no": "1Z999"}.

Input.

Component	Value
Sink	Redis hash per order
Delta components	updatedFields, removedFields, truncatedArrays
Nested paths	shipping.tracking_no → single Redis field shipping.tracking_no
Change stream option	full_document=None

Code.

import redis
from pymongo import MongoClient

r      = redis.Redis(host="redis.internal", port=6379)
client = MongoClient("mongodb://mongos.internal:27017/")
coll   = client.shop.orders

def apply_delta(doc_id, ud):
    key = f"order:{doc_id}"
    p = r.pipeline()

    # updatedFields → HSET
    if ud.get("updatedFields"):
        # updatedFields is a dict; flatten values that are BSON-native
        pairs = {}
        for path, value in ud["updatedFields"].items():
            pairs[path] = str(value) if not isinstance(value, str) else value
        p.hset(key, mapping=pairs)

    # removedFields → HDEL
    if ud.get("removedFields"):
        p.hdel(key, *ud["removedFields"])

    # truncatedArrays → mark newSize on the array field
    for trunc in ud.get("truncatedArrays", []):
        p.hset(key, f"{trunc['field']}.length", trunc["newSize"])

    p.execute()

with coll.watch(full_document=None) as stream:
    for change in stream:
        op = change["operationType"]
        doc_id = change["documentKey"]["_id"]

        if op == "insert" or op == "replace":
            # For insert/replace, delta model degenerates — dump the whole doc as HSET
            flat = flatten(change["fullDocument"])   # dotted paths
            r.hset(f"order:{doc_id}", mapping=flat)
        elif op == "update":
            apply_delta(doc_id, change["updateDescription"])
        elif op == "delete":
            r.delete(f"order:{doc_id}")

def flatten(doc, prefix=""):
    out = {}
    for k, v in doc.items():
        path = f"{prefix}{k}"
        if isinstance(v, dict):
            out.update(flatten(v, prefix=path + "."))
        else:
            out[path] = str(v)
    return out

Step-by-step explanation.

1. updateDescription.updatedFields maps directly to Redis HSET field-value pairs. Nested paths like "shipping.tracking_no" are preserved verbatim — Redis hashes have flat field names, so the dotted path becomes the field name. Consumers reading the hash later split on . to reconstruct nested structure.
2. updateDescription.removedFields is a list of paths to HDEL. Same dotted-path convention.
3. truncatedArrays is subtler — the delta doesn't include the truncated element values, just the new size. The consumer records the new length and can iterate field.0, field.1, ..., field.<newSize-1> if it needs the surviving elements. If the array truncated to size 0, the consumer can safely delete every field.* entry.
4. insert and replace events degenerate to "flatten the whole document and HSET." This branch is where updateLookup would be free — the event already carries fullDocument — so leaving full_document=None costs nothing here.
5. delete is a single r.delete(key). The documentKey._id is all the router needs.

Output.

Op	Redis command	Extra Mongo reads	Notes
insert	HSET all fields	0	fullDocument on event
replace	HSET all fields	0	fullDocument on event
update	HSET updated + HDEL removed	0	delta only
delete	DEL key	0	documentKey._id sufficient

Rule of thumb. For delta-apply sinks (Redis hashes, in-memory materialised views, aggregate counters), leave full_document=None. The delta is the payload and enabling updateLookup just adds primary-side I/O with no consumer-side use.

Worked example — pre-image capture for audit logging

Detailed explanation. An audit-log sink needs both the before and after state of every update, plus the deleted document for every delete. Mongo 6.0+ supports pre-image capture: enable it on the collection, then use fullDocumentBeforeChange alongside fullDocument on the watch. Pre-image capture is an expensive collection-level setting — copy every pre-image to an internal collection, capped by TTL — but for audit workloads it's the only path to a complete before/after record.

• Enable pre-image capture. collMod with changeStreamPreAndPostImages.
• Watch options. full_document="required" + full_document_before_change="required".
• Cost. Storage grows proportional to write rate × document size × retention window.

Question. Configure pre-image capture on shop.orders and build an audit consumer that emits {before, after, op, ts} records to a compliance log.

Input.

Component	Value
Collection	shop.orders
Pre-image retention	7 days
Watch options	full_document=required, full_document_before_change=required
Audit sink	S3 append-only bucket

Code.

// Enable pre-image capture on the collection (mongosh)
db.runCommand({
  collMod: "orders",
  changeStreamPreAndPostImages: { enabled: true },
});

// Optionally set retention on the internal pre-image collection
db.adminCommand({
  setClusterParameter: {
    changeStreamOptions: {
      preAndPostImages: { expireAfterSeconds: 604800 },   // 7 days
    },
  },
});

from pymongo import MongoClient
import json, boto3

s3 = boto3.client("s3")
client = MongoClient("mongodb://mongos.internal:27017/")
coll   = client.shop.orders

def emit(record):
    key = f"audit/{record['ts']}/{record['id']}.json"
    s3.put_object(Bucket="compliance-logs", Key=key,
                  Body=json.dumps(record).encode())

with coll.watch(
    full_document="required",
    full_document_before_change="required",
) as stream:
    for change in stream:
        record = {
            "op":     change["operationType"],
            "id":     str(change["documentKey"]["_id"]),
            "before": change.get("fullDocumentBeforeChange"),
            "after":  change.get("fullDocument"),
            "ts":     str(change["clusterTime"]),
        }
        emit(record)
        save_resume_token(change["_id"])

Step-by-step explanation.

1. The collMod command enables pre-image capture on the orders collection. Every subsequent update and delete copies the pre-image to an internal collection (config.system.preimages). Pre-existing documents are not retroactively captured — only mutations from the enable-time forward.
2. The cluster-parameter setting bounds the retention. Pre-images older than 7 days are pruned; consumers that fall behind more than 7 days will hit PreImageMissingError on full_document_before_change="required" events.
3. The watch is opened with both full_document="required" and full_document_before_change="required". required means "error if the pre or post image is unavailable." For an audit workload that must never lose a before/after record, this is the correct posture — errors surface loudly instead of silently dropping.
4. Every event is written to S3 as an append-only record with the before/after state, op, id, and cluster time. Consumers of the audit log get a complete replayable history of every mutation.
5. Note: pre-image capture roughly doubles the write amplification on the source collection. It is expensive. Only enable it for collections that need it (audit, GDPR, financial compliance) — not by default.

Output.

Property	Value
Pre-image storage	~1x document size × write rate × retention
Extra oplog cost	~0 (pre-image stored separately)
Audit-log completeness	before + after on every update/delete
Consumer error mode	PreImageMissingError on retention gap
Suitable for	audit, GDPR, replay, financial compliance

Rule of thumb. Pre-image capture is expensive. Enable it only on the specific collections that need before/after records for compliance or audit. Do not enable it "just in case" — the storage growth is proportional to the write rate and will surprise you.

Senior interview question on full document lookup

A senior interviewer might ask: "You have three downstream sinks: a Snowflake warehouse, a Redis materialised view, and a compliance audit log. Walk me through the fullDocument and fullDocumentBeforeChange settings for each, the cost model, and what you'd do if a single change stream had to feed all three."

Solution Using per-sink watch options with a fan-out router

# fanout_solution.py — three sinks, three watch configurations, one router
from pymongo import MongoClient
import logging

log = logging.getLogger("cdc.fanout")

def run_warehouse_sink(client, coll):
    # Warehouse wants current row → updateLookup
    with coll.watch(
        pipeline=[{"$match": {"operationType": {"$in":
            ["insert", "update", "replace", "delete"]}}}],
        full_document="updateLookup",
        resume_after=load_token("warehouse"),
    ) as stream:
        for change in stream:
            snowflake_upsert(change.get("fullDocument")
                             or change["documentKey"])
            save_token("warehouse", change["_id"])

def run_redis_view_sink(client, coll):
    # Materialised view applies deltas → default (no updateLookup)
    with coll.watch(
        full_document=None,
        resume_after=load_token("redis"),
    ) as stream:
        for change in stream:
            apply_redis_delta(change)
            save_token("redis", change["_id"])

def run_audit_sink(client, coll):
    # Audit wants before/after → pre-image capture required at collection level
    with coll.watch(
        full_document="required",
        full_document_before_change="required",
        resume_after=load_token("audit"),
    ) as stream:
        for change in stream:
            audit_write({
                "op":     change["operationType"],
                "id":     str(change["documentKey"]["_id"]),
                "before": change.get("fullDocumentBeforeChange"),
                "after":  change.get("fullDocument"),
                "ts":     str(change["clusterTime"]),
            })
            save_token("audit", change["_id"])

Step-by-step trace.

Sink	Watch option	Extra Mongo I/O	Payload size	Notes
Snowflake warehouse	updateLookup	1 read per update	delta + fullDocument	current row available
Redis materialised view	none	0	delta only	delta apply
Audit log	required + pre-image required	pre-image collection writes	delta + before + after	compliance-grade

Each sink runs its own consumer process with its own resume token. The three watches are independent — they don't share state at the source or the sink. Total load on the Mongo primary is one updateLookup per update (from the warehouse consumer) plus one pre-image write per update (from the audit consumer's requirement); the Redis consumer adds zero. The trade-offs are honest and per-sink — no over-provisioning.

Output:

Sink	Round-trips/event	Storage impact	Latency impact
Warehouse	1 (updateLookup on server)	none	small (+1-2 ms)
Redis view	0	none	none
Audit	0 (pre-image cached)	double writes on source	small (+1 ms)

Why this works — concept by concept:

• Per-consumer watch options — each sink opens its own watch with the options it needs. No "one size fits all"; costs are billed only to the consumers that need them.
• updateLookup for hydrating sinks — moves the extra read from consumer to server. One round-trip becomes zero; primary I/O is unchanged.
• default for delta-apply sinks — Redis view consumes deltas natively; enabling updateLookup would cost primary I/O with zero benefit.
• pre-image capture for audit — the only path to a complete before/after record. Expensive but bounded — enable only on collections needing it, cap retention with expireAfterSeconds.
• Independent resume tokens — three sinks means three checkpoints. A slow sink can fall behind without affecting others.
• Cost — total Mongo primary load is bounded by (updateLookup rate) + (pre-image write rate). Per-sink CPU/network is bounded by (event rate × option cost). Big-O is O(events) per sink at runtime; O(consumers × collections) at watch-open time.

ETL
Topic — etl
ETL problems on full-document hydration and delta apply

Practice →

Optimization Topic — optimization Optimization problems on hydration cost trade-offs

Practice →

---

5. Cluster-wide + sharded change streams + Debezium

mongos-hosted cluster-wide watches are the production shape; Debezium is the Kafka-Connect wrapper you graduate to when the team lives in Kafka Connect

The mental model in one line: for single-collection ingest, open collection.watch(...) against a replica set; for multi-collection ingest, open db.watch(...) or client.watch(...) against mongos in a sharded cluster; for Kafka-native shops that want offset persistence and connect-plane operations, run the Debezium MongoDB source connector, which wraps the same change-stream API and stores resume tokens in the Kafka offsets topic. The choice is about team topology and existing operational infrastructure, not about capability — every path reads the same underlying oplog.

The three watch scopes revisited.

• collection.watch(...). Single collection. Simplest. Cursor lives on the primary of the replica set (or on the shard primary if the collection is on one shard).
• db.watch(...). All collections in one database. Cursor lives on the primary of the replica set (unsharded) or on mongos (sharded).
• client.watch(...). Entire cluster. Cursor lives on mongos (sharded) or on the replica set primary (unsharded). Sees every event across every db and collection except a small set of system collections.

Sharded watch behaviour on mongos.

• Fan-out. mongos opens sub-cursors on every shard's oplog and merges the events by cluster time.
• Ordering. The merged stream is globally ordered by cluster time. Two events on different shards with the same cluster-time ordinal have implementation-defined tie-breaking, but total ordering is preserved.
• Latency. The merger waits for the slowest shard — if one shard falls behind, the whole stream stalls. Watch shard-level oplog lag as an SLI.
• Failover. When a shard's primary fails over, mongos transparently re-opens the sub-cursor on the new primary. The consumer sees a brief pause, then a resumption.

The Debezium Mongo connector at a glance.

• What it is. A Kafka-Connect source connector that opens a change stream against Mongo, transforms events into Debezium-style envelopes, and produces to Kafka topics named {prefix}.{db}.{coll} by default.
• How it stores resume tokens. In the Kafka Connect offsets topic, keyed by connector name + stream identifier. Restart-safe out of the box.
• What it adds. Schema-registry integration (Avro / Protobuf), SMT (Single Message Transforms) pipeline, dead-letter queue support, operational APIs via Kafka Connect REST.
• What it costs. You now run Kafka Connect. Ops burden is real if the team wasn't already there.

Debezium event envelope shape.

• Key. {"_id": {"$oid": "..."}} — the document _id. Ensures per-document ordering via Kafka partitioning.
• Value. {"op": "c|u|d|r", "ts_ms": ..., "before": {...}, "after": {...}, "source": {...}, "transaction": {...}}. c = create (insert), u = update, d = delete, r = read (snapshot).
• Compare vs raw change-stream event. Debezium normalises across databases (Mongo, Postgres, MySQL) so downstream consumers can be database-agnostic. Raw change-stream events are Mongo-specific and richer.

Native driver cursor vs Debezium — the decision framework.

• Native driver. 30-60 lines of Python/Node, one process, direct control. Best for small teams, single-purpose ingest, prototypes, and when you don't already run Kafka Connect.
• Debezium. Zero code (config only), integrates with existing Kafka Connect cluster, schema registry, SMTs, DLQ. Best for Kafka-native shops with multiple CDC sources, teams that value config-over-code, and multi-database CDC unification.
• Migration. From native to Debezium: rewrite as connector config, cutover with careful offset alignment. From Debezium to native: extract logic, run in parallel until confidence.

Common interview probes on cluster-wide + Debezium.

• "What runs the watch cursor in a sharded cluster?" — mongos.
• "How does mongos handle a slow shard?" — waits for the slowest; stream stalls if the shard falls further behind.
• "When would you pick Debezium over the native driver?" — Kafka Connect already in production, multi-database CDC, config-over-code preference.
• "How does Debezium persist resume tokens?" — in the Kafka Connect offsets topic.

Worked example — cluster-wide watch feeding per-collection Kafka topics

Detailed explanation. A production ingest service watches an entire sharded cluster and produces one Kafka topic per collection. mongos handles the fan-out and merge; the consumer sees a single globally-ordered stream. This is the canonical "single cursor, many topics" topology that beats "one cursor per collection" for anything up to a few thousand events/sec across a few dozen collections.

• Deployment. Sharded cluster, 3 shards, mongos endpoint.
• Topology. One consumer process, one client.watch(...), N Kafka topics.
• Ordering. Globally ordered by cluster time via mongos merge.

Question. Implement the ingest service. Configure server-side filtering to drop system collections. Show how the Kafka key preserves per-document ordering. Discuss what happens when one shard falls behind.

Input.

Component	Value
Deployment	Mongo sharded cluster, 3 shards
Endpoint	mongos.internal:27017
Watch scope	cluster-wide
Sink	Kafka, one topic per (db, coll)
System dbs to drop	admin, local, config
Message key	documentKey._id (string)

Code.

from pymongo import MongoClient
from kafka import KafkaProducer
import json, logging

log = logging.getLogger("cdc.cluster")

client = MongoClient(
    "mongodb://mongos.internal:27017/",
    appname="cluster-cdc-v1",
)
producer = KafkaProducer(
    bootstrap_servers="kafka.internal:9092",
    value_serializer=lambda v: json.dumps(v, default=str).encode(),
    key_serializer=lambda k: k.encode(),
    linger_ms=20, acks="all",
)

pipeline = [
    {"$match": {"ns.db": {"$nin": ["admin", "local", "config"]}}},
]

with client.watch(pipeline=pipeline, full_document="updateLookup") as stream:
    for change in stream:
        db_name   = change["ns"]["db"]
        coll_name = change["ns"].get("coll")
        if not coll_name:
            continue
        topic = f"mongo.{db_name}.{coll_name}"
        key   = str(change["documentKey"]["_id"])
        producer.send(topic, key=key, value=change)
        producer.flush()
        save_resume_token(change["_id"])

Topology diagram
================

  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐
  │   shard 0   │  │   shard 1   │  │   shard 2   │
  │  oplog rs0  │  │  oplog rs1  │  │  oplog rs2  │
  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘
         │                │                │
         └────────┬───────┴────────┬───────┘
                  ▼                ▼
              ┌──────────────────────┐
              │        mongos        │  ← client.watch() opens here
              │  merge by clusterTime│
              └──────────┬───────────┘
                         ▼
                 change-stream events
                         ▼
                 ┌────────────────┐
                 │  ingest.py     │
                 └────────┬───────┘
                          ▼
              ┌──────────────────────┐
              │ Kafka: N topics      │
              │ mongo.<db>.<coll>    │
              └──────────────────────┘

Step-by-step explanation.

1. client.watch(...) opens a cluster-wide cursor against mongos. Under the covers, mongos opens per-shard cursors on each shard's oplog and merges the events by cluster time. The consumer sees one globally-ordered stream.
2. The server-side $match filter drops system-db events before they cross the wire. On a busy deployment, this alone can cut wire bytes by 70%+ because internal Mongo housekeeping generates a steady stream of admin.system.* events.
3. full_document="updateLookup" is applied for downstream sinks that want the current row. For a Kafka topic that fans out to multiple downstream consumers with different needs, enabling updateLookup at the source is the pragmatic default — the event carries the doc, and downstream can ignore what it doesn't want.
4. The Kafka message key is documentKey._id (serialized to string). This guarantees per-document ordering: all events for the same _id land on the same Kafka partition, and Kafka preserves order within a partition. Downstream consumers doing state derivation see a consistent per-document sequence.
5. When one shard falls behind (say shard 2's oplog lag jumps to 30 seconds), mongos cannot advance the merged stream past the shard-2 lag — it has to wait for shard-2 events with cluster time earlier than events already-available from shards 0 and 1 to preserve global ordering. The whole stream stalls. Alert on per-shard oplog lag to catch this before consumers pile up.

Output.

Metric	Value
Cursors open	1 (cluster-wide, hosted on mongos)
Under the covers	3 sub-cursors (one per shard)
Topics fed	N per (db, coll)
Ordering	globally ordered by cluster time
Per-document ordering on Kafka	preserved (key = _id)
Slow-shard failure mode	stream stalls; alert on per-shard oplog lag

Rule of thumb. For cluster-wide ingest, run one consumer with one client.watch(...) against mongos. Alert on per-shard oplog lag as an SLI — a stalled stream is the symptom of a stuck shard, and shard-side fixes are always in a different team's runbook.

Worked example — Debezium Mongo source connector config

Detailed explanation. A Kafka-native shop with an existing Kafka Connect cluster deploys the Debezium Mongo source connector. Config is JSON posted to the Connect REST API. The connector handles resume-token persistence, schema registration, and event emission — the operational surface is curl commands against Connect. Compare against the native-driver equivalent.

• Connector class. io.debezium.connector.mongodb.MongoDbConnector.
• Topic naming. {topic.prefix}.{db}.{coll}.
• Offset storage. Connect offsets topic (transparent).
• Schema. Avro or JSON Schema, integrated with Schema Registry.

Question. Write the Debezium Mongo connector JSON for the same shop.orders workload. Explain each key and how it maps to the native-driver equivalents from earlier sections.

Input.

Debezium key	Native-driver equivalent	Purpose
mongodb.connection.string	MongoClient URI	Where to connect
topic.prefix	topic-name template	Kafka topic naming
capture.mode	full_document option	delta / lookup / pre-image
collection.include.list	pipeline $match on ns	filter
snapshot.mode	bootstrap policy	initial snapshot behaviour

Code.

{
  "name": "mongo-shop-source",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "tasks.max": "1",

    "mongodb.connection.string": "mongodb://mongos.internal:27017/?replicaSet=rs0",
    "topic.prefix": "mongo",

    "capture.mode": "change_streams_update_full_with_pre_image",
    "capture.scope": "database",
    "capture.target": "shop",

    "collection.include.list": "shop.orders,shop.customers,shop.products",

    "snapshot.mode": "initial",
    "snapshot.max.threads": "2",

    "poll.interval.ms": "500",

    "cursor.max.await.time.ms": "1000",

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry.internal:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry.internal:8081",

    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "mongo-shop-dlq",
    "errors.deadletterqueue.context.headers.enable": "true"
  }
}

# Deploy via Kafka Connect REST
curl -X POST -H "Content-Type: application/json" \
  --data @mongo-shop-source.json \
  http://kafka-connect.internal:8083/connectors

# Check status
curl http://kafka-connect.internal:8083/connectors/mongo-shop-source/status

Step-by-step explanation.

1. mongodb.connection.string points at mongos — same target as the native MongoClient(...) URI. The connector opens cluster-wide change streams by default; scoping is controlled by capture.scope and capture.target.
2. topic.prefix = "mongo" gives topic names like mongo.shop.orders. Matches the native-driver convention chosen earlier; downstream Kafka consumers are indifferent to which source populated the topic.
3. capture.mode = "change_streams_update_full_with_pre_image" maps to full_document="required" + full_document_before_change="required" in the native driver. It's the audit-grade mode: full pre and post images on every update. Cheaper modes are change_streams (delta only) and change_streams_update_full (post-image only).
4. snapshot.mode = "initial" bootstraps the sink from a snapshot on first startup, exactly like the bootstrap function shown in Section 3. Alternatives: never (no snapshot; risky), when_needed (snapshot on history-lost recovery).
5. Kafka Connect handles offset storage transparently — no code, no explicit token save. The offsets topic in Connect stores the resume tokens keyed by connector name. Restarting the connector picks up at the last committed offset. Downstream schemas are registered with Schema Registry via the Avro converters, giving downstream consumers strong-typed access.

Output.

Layer	Native driver	Debezium
Code	~30-100 lines	JSON config
Resume-token storage	your choice (Mongo, disk)	Kafka Connect offsets
Schema	ad-hoc JSON	Avro + Schema Registry
Restart	reload token, reopen watch	Connect restarts task
DLQ	build yourself	built in
Ops surface	run your own process	Kafka Connect REST

Rule of thumb. If your team already runs Kafka Connect, ship Debezium. The connect-plane ergonomics (config-driven, REST-managed, DLQ built-in, schema-registry integrated) are worth the migration cost. If your team does not run Kafka Connect, do not adopt it just for Mongo CDC — the native driver is 30 lines and one process.

Worked example — native cursor vs Debezium decision matrix

Detailed explanation. The last part of most senior interviews on this topic is a "which would you pick?" question. Walk through a decision matrix that scores the two options across five axes: existing infrastructure, team size, operational maturity, schema needs, and future roadmap.

• Existing infra. Kafka Connect already running? Push to Debezium.
• Team size. Small team, 1-2 CDC pipelines? Native is simpler.
• Ops maturity. New team without Connect ops experience? Native.
• Schema needs. Schema Registry integration valued? Debezium.
• Roadmap. Multi-database CDC (Postgres + Mongo + MySQL)? Debezium.

Question. Score two hypothetical teams — Team A (small, single-Mongo, no Kafka) and Team B (large, multi-source CDC, established Kafka Connect) — on all five axes and recommend the tool for each.

Input.

Axis	Team A	Team B
Existing Kafka Connect	no	yes
Team size	3 engineers	30 engineers
Ops maturity for Connect	low	high
Schema Registry in stack	no	yes
Multi-database CDC roadmap	no	yes

Code.

# decision.py — score the axes and pick
def score_tool(team, tool):
    weights = {
        "existing_connect": 3,
        "team_size":       -2,   # bigger team → more overhead OK
        "ops_maturity":     2,
        "schema_registry":  2,
        "multi_source":     3,
    }
    axis_scores = {
        "existing_connect": {"native": 0, "debezium": 3 if team["existing_connect"] else -3},
        "team_size":       {"native": 3 if team["team_size"] < 10 else 1, "debezium": 1 if team["team_size"] < 10 else 3},
        "ops_maturity":    {"native": 2, "debezium": 3 if team["ops_maturity"] == "high" else -2},
        "schema_registry": {"native": 0, "debezium": 2 if team["schema_registry"] else 0},
        "multi_source":    {"native": -2 if team["multi_source"] else 0, "debezium": 3 if team["multi_source"] else 0},
    }
    total = sum(axis_scores[k][tool] for k in weights)
    return total

team_a = {"existing_connect": False, "team_size": 3, "ops_maturity": "low",
          "schema_registry": False, "multi_source": False}
team_b = {"existing_connect": True,  "team_size": 30, "ops_maturity": "high",
          "schema_registry": True,   "multi_source": True}

for label, team in [("Team A", team_a), ("Team B", team_b)]:
    native_s   = score_tool(team, "native")
    debezium_s = score_tool(team, "debezium")
    winner = "native" if native_s > debezium_s else "debezium"
    print(f"{label}: native={native_s} debezium={debezium_s} → {winner}")

Team A: native=5 debezium=-4 → native
Team B: native=1 debezium=14 → debezium

Step-by-step explanation.

1. Team A has no Kafka Connect, a small team, and no multi-source roadmap. The native driver is 30 lines; deploying Debezium would require standing up Kafka Connect just for one CDC pipeline. Ops burden dwarfs the code savings. Pick native.
2. Team B has an established Connect cluster, a large team already operating multiple connectors, and a roadmap for Postgres + Mongo + MySQL CDC. Adding one more Debezium connector is config-only; standardising on Debezium across all three databases unifies the operational surface. Pick Debezium.
3. The score model is illustrative — real teams weight axes differently. The point is not the exact score but the framework: existing infra × team size × ops maturity × schema needs × roadmap.
4. The migration cost between the two is real but manageable. Native → Debezium: rewrite as connector config, cut over with careful offset alignment. Debezium → native: extract the transformation logic, run parallel until confidence. Neither is a one-way door.
5. The senior interview signal is naming the axes and defending the ranking, not picking a tool. Interviewers reward candidates who can articulate why one wins for a given team; they downgrade candidates who default to whichever tool they last used.

Output.

Team	Recommendation	Reasoning
Team A (small, no Connect)	Native driver	30 lines vs a whole Connect deployment
Team B (large, multi-source)	Debezium	Config-only in existing Connect; schema unified

Rule of thumb. Pick Debezium if you already run Kafka Connect and want config-over-code plus schema-registry integration. Pick the native driver if you don't run Connect and don't want to. Never pick a tool "because it's modern" — the ops burden is always more than the code burden.

Senior interview question on sharded change streams and Debezium

A senior interviewer might ask: "Design the Mongo CDC pipeline for a sharded cluster feeding Kafka. Walk me through where the cursor lives, what ordering guarantees you get, which watch scope you pick, and whether you'd use the native driver or Debezium — and what would push you from one to the other over time."

Solution Using a mongos-hosted cluster-wide watch with a native-then-Debezium graduation path

# Phase 1 — native driver (day 1 to month 6)
# ingest_v1.py — cluster-wide watch on mongos, per-collection Kafka topics
from pymongo import MongoClient
from kafka import KafkaProducer
import json

client = MongoClient("mongodb://mongos.internal:27017/", appname="mongo-cdc-v1")
producer = KafkaProducer(
    bootstrap_servers="kafka.internal:9092",
    value_serializer=lambda v: json.dumps(v, default=str).encode(),
    key_serializer=lambda k: k.encode(),
    acks="all", linger_ms=20,
)

pipeline = [
    {"$match": {"ns.db": {"$nin": ["admin", "local", "config"]},
                "operationType": {"$in":
                    ["insert", "update", "replace", "delete"]}}},
]

with client.watch(pipeline=pipeline, full_document="updateLookup") as stream:
    for change in stream:
        topic = f"mongo.{change['ns']['db']}.{change['ns'].get('coll')}"
        key   = str(change["documentKey"]["_id"])
        producer.send(topic, key=key, value=change)
        producer.flush()
        save_resume_token(change["_id"])

// Phase 2 — Debezium migration when Kafka Connect is available (month 6+)
// mongo-shop-source.json — same topic layout, same ordering, offsets in Connect
{
  "name": "mongo-shop-source",
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
    "tasks.max": "1",
    "mongodb.connection.string": "mongodb://mongos.internal:27017/",
    "topic.prefix": "mongo",
    "capture.mode": "change_streams_update_full",
    "collection.include.list": "shop.orders,shop.customers,shop.products,billing.invoices",
    "snapshot.mode": "initial",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry.internal:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry.internal:8081",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "mongo-shop-dlq"
  }
}

Step-by-step trace.

Phase	Component	Detail
Cursor host	mongos	fans across shards; merges by cluster time
Watch scope	cluster-wide	one cursor, N collections
Filter	server-side $match	drop system dbs; keep CRUD ops
Hydration	updateLookup	current row on every update event
Topic naming	mongo.<db>.<coll>	matches Debezium convention
Kafka key	documentKey._id	per-document ordering preserved
Phase 1 resume	Mongo checkpoint coll	one row per stream
Phase 2 resume	Kafka Connect offsets	transparent to operators
Migration trigger	Kafka Connect adopted for other sources	consolidation win
Migration cost	~1 week to align offsets, cutover, decom native	manageable

Phase 1 ships a native-driver ingest in ~50 lines. It carries the workload for months. When the platform team stands up Kafka Connect (typically driven by Postgres CDC via Debezium Postgres), the Mongo CDC pipeline migrates to Debezium in Phase 2 — same topic names, same key strategy, same ordering guarantee. Downstream Kafka consumers see zero change. Total operational surface consolidates onto Kafka Connect.

Output:

Metric	Phase 1 (native)	Phase 2 (Debezium)
Ingest processes to operate	1 custom	1 Connect task (shared cluster)
Resume-token storage	Mongo checkpoint coll	Connect offsets topic
Schema management	ad-hoc JSON	Avro + Schema Registry
DLQ	build yourself	built-in
Migration risk	—	low (same topic layout)
Downstream impact	—	zero

Why this works — concept by concept:

• Mongos-hosted cluster-wide watch — one cursor, N collections, globally ordered. Fan-out is the server's job; consumer sees a single stream. No app-side shard coordination.
• Server-side filter — drops system-db events before they cross the wire. Every byte saved is CPU saved on the consumer side.
• updateLookup at source — hydrates events once. Downstream Kafka consumers get the current row without extra Mongo round-trips.
• Kafka key = documentKey._id — preserves per-document ordering end-to-end. Consumers doing state derivation see a coherent per-document sequence.
• Native-first, Debezium-later — matches capability to team maturity. Phase 1 ships in a week; Phase 2 consolidates when the Connect infrastructure exists. No sunk cost in either direction.
• Cost — Phase 1 costs one process to operate. Phase 2 amortises across an existing Connect cluster. Big-O is O(events) for both at runtime; O(1) for operational headcount because Debezium reuses existing infra.

Streaming
Topic — streaming
Streaming problems on sharded ingest and topology decisions

Practice →

ETL
Topic — etl
ETL problems on connector vs custom-driver trade-offs

Practice →

---

Cheat sheet — MongoDB change stream recipes

Basic collection.watch() with resumeAfter

• Pattern. Single-collection cursor, restart-safe from a saved token. The 30-line default for one-collection ingest.
• Code shape. with coll.watch(resume_after=token, full_document='updateLookup') as stream: inside a while True: loop that reloads the token on transient errors.
• When to use. Simplest possible ingest, one collection, one sink. Move to db.watch(...) or client.watch(...) when a second collection joins the party.
• Gotcha. resumeAfter errors on invalidate; prefer startAfter on Mongo 4.2+ so drop-and-recreate migrations don't force a bootstrap.

fullDocument: 'updateLookup' for hydrated updates

• Pattern. Server round-trips to the primary after each update and includes the current document on the change event. Consumer avoids its own lookup.
• Code shape. coll.watch(full_document='updateLookup'). In Node/JS drivers, collection.watch([], { fullDocument: 'updateLookup' }).
• When to use. Warehouse sinks, search-index sinks, any sink that upserts the current row. Do not use for materialised views that apply deltas.
• Gotcha. Not point-in-time — the lookup returns the current document, potentially reflecting later updates. Practically consistent within a few ms.

Cluster-wide client.watch() with server-side filter

• Pattern. One cursor covers every non-system collection. $match drops the noise before it hits the wire.
• Code shape. client.watch(pipeline=[{'$match': {'ns.db': {'$nin': ['admin','local','config']}, 'operationType': {'$in': ['insert','update','replace','delete']}}}], full_document='updateLookup').
• When to use. Multi-collection ingest against a sharded cluster, up to a few thousand events/sec across a few dozen collections.
• Gotcha. Merger stalls on the slowest shard — monitor per-shard oplog lag as an SLI, not just consumer lag.

Debezium Mongo connector JSON

• Pattern. Kafka-Connect-hosted CDC. Config-only. Offsets in the Connect offsets topic. Schema via Registry.
• Code shape. POST /connectors with connector.class = io.debezium.connector.mongodb.MongoDbConnector, mongodb.connection.string = ..., topic.prefix = mongo, capture.mode = change_streams_update_full, snapshot.mode = initial.
• When to use. Kafka Connect already in the stack; multi-source CDC (Postgres + Mongo); schema-registry-first team.
• Gotcha. Do not stand up Kafka Connect just for Mongo CDC — the native driver is 30 lines and one process. Adopt Debezium when Connect exists for other reasons.

Resume-token checkpoint schema

• Pattern. One document per stream in a dedicated Mongo checkpoint collection. Persist after every downstream ack.
• Code shape. {_id: "<stream-id>", token: <opaque BSON>, start_at_operation_time: <Timestamp | null>, saved_at: <unix seconds>} in cdc_ops.resume_tokens.
• When to use. Native-driver ingest. Kafka Connect stores tokens in the offsets topic automatically.
• Gotcha. Persist after the sink ack for at-least-once. Persist before the ack for at-most-once (drops on crash — rarely what you want).

Frequently asked questions

What are MongoDB change streams and why did they replace ad-hoc oplog tailers?

mongodb change streams are the official cursor API layered on top of the replica-set oplog. A single collection.watch(...), db.watch(...), or client.watch(...) call returns a stable, documented, driver-supported stream of change events, each carrying _id (resume token), operationType, ns, documentKey, updateDescription, optional fullDocument, and clusterTime. Before change streams (added in Mongo 3.6), CDC out of Mongo meant tailing local.oplog.rs with a tailable cursor, decoding Mongo-internal document formats yourself, hand-rolling resume logic against per-shard Timestamp markers, and re-implementing every edge case (filtering, sharded coordination, delete hydration) as application code. That approach worked but was ~300 lines per pipeline, silently broke on minor Mongo version bumps, and could not watch a sharded cluster as a single ordered stream. Change streams collapse that entire pile into a documented API — server-side aggregation-pipeline filtering, opaque resume tokens, cluster-wide fan-out via mongos, and per-consumer fullDocument hydration — that ships in every official driver.

What is a resume token and how do I use it correctly?

A resume token is an opaque BSON document ({"_data": "826..."}) that encodes a position in the oplog — plus cluster time in sharded deployments. The driver returns one on every change event as _id. You never look inside; you only store it and pass it back to resume_after (or start_after on 4.2+) when re-opening the stream. The single most-interviewed detail: persist the token after the downstream sink acknowledges the event, not before. This gives at-least-once semantics — a crash between the sink ack and the token save replays one event, which an idempotent sink handles cleanly. Persisting before the ack gives at-most-once, which drops events on crashes. Store the token in a Mongo checkpoint collection, a Kafka offsets topic (for Debezium), or an external KV store. If the oplog rolls past the position the token references, the next resumeAfter call errors with ChangeStreamHistoryLost (code 286) — the recovery path bootstraps the sink from a snapshot and reopens the stream with startAtOperationTime = current cluster time.

Do I need Debezium or can I just use the native driver?

Both work. Native driver. 30–60 lines of Python, Node, Java, Go, or Rust — one process, direct control over the loop, resume-token storage of your choice. Best for small teams, one or two CDC pipelines, prototypes, and any shop that does not already run Kafka Connect. Debezium. Zero code (JSON config posted to Kafka Connect REST), transparent offset persistence in the Connect offsets topic, native Schema Registry integration, built-in DLQ, config-driven SMTs. Best for Kafka-native shops with multi-source CDC (Postgres + Mongo + MySQL), teams that value config-over-code, and platforms that already operate a Connect cluster. The decision matrix. Existing Kafka Connect? Push to Debezium. Small team, single Mongo source? Native is simpler. Do not stand up Kafka Connect just for one Mongo CDC pipeline — the ops burden dwarfs the code burden. Do not stick with the native driver "because we always have" if a Connect cluster is already running for other CDC sources. The senior signal in an interview is naming the decision axes (existing infra, team size, ops maturity, schema needs, multi-source roadmap) and defending the ranking for the specific shop.

What is fullDocument: 'updateLookup' and when do I use it?

By default, a change-stream update event carries only the delta — updateDescription: {updatedFields, removedFields, truncatedArrays}. If the downstream sink needs the current document (a warehouse upserting the current row, a search index reindexing, a Kafka topic feeding heterogeneous downstream consumers), setting full_document='updateLookup' on the watch tells the server to do a follow-up read against the primary and include the current state of the document in the event as fullDocument. The cost. One extra Mongo primary read per update event, plus a slightly larger event payload. On a 1000-event/sec stream with 500-byte documents, that is 500 KB/sec of extra wire bytes plus 1000 extra primary reads/sec. The gotcha. updateLookup is not point-in-time — it returns the document as it exists at read time, which may already reflect later updates. For most sinks that upsert the current row, this is exactly what you want. For sinks that maintain a materialised view by applying deltas, leave it off — the delta is the payload and the extra hydration cost buys nothing. Mongo 6.0+ adds required (errors if pre/post images unavailable) and whenAvailable (best-effort with pre-image capture) for audit-grade workloads.

Can I watch an entire cluster with one cursor?

Yes. client.watch(pipeline, full_document='updateLookup') opens a cluster-wide watch that covers every non-system database and collection. In a sharded deployment, the cursor lives on mongos, which opens per-shard sub-cursors on each shard's oplog and merges the events by cluster time — the consumer sees a single globally-ordered stream. This is the canonical topology for multi-collection ingest with modest event rates (up to a few thousand events/sec across a few dozen collections). Push filtering into the aggregation pipeline ([{$match: {'ns.db': {$nin: ['admin','local','config']}, 'operationType': {$in: ['insert','update','replace','delete']}}}]) so system-database noise and uninteresting op types are dropped server-side. Route each event by ns.db and ns.coll into per-collection Kafka topics or downstream sinks. Beyond a few thousand events/sec or a few dozen collections, shard the ingest by opening separate db.watch(...) or collection.watch(...) cursors on independent consumers — each with its own resume token — to bound the blast radius of a slow consumer.

How do change streams work in a sharded cluster and what breaks when a shard falls behind?

In a sharded cluster, change streams open against mongos, which coordinates the fan-out and merge. mongos opens per-shard sub-cursors on each shard's oplog, receives events tagged with cluster time, and interleaves them into a single globally-ordered stream on the way back to the client. Ordering guarantee: events are returned in cluster-time order regardless of which shard produced them; tie-breaking within the same cluster-time ordinal is implementation-defined but consistent within a run. The catch. The merger cannot advance past the slowest shard. If shard 2's oplog lag spikes to 30 seconds (say a large index build or a stalled secondary), mongos has to wait for shard-2 events with cluster time earlier than events already-available from shards 0 and 1 to preserve global ordering. The whole stream stalls until shard 2 catches up. This is why per-shard oplog lag is a first-class SLI in production Mongo CDC pipelines — alerting only on consumer lag misses the root cause. Debezium and native cursors both surface this failure the same way (stalled cursor, no events flowing); the fix is always at the shard, not at the consumer. When a shard's primary fails over, mongos transparently re-opens the sub-cursor on the new primary — the consumer sees a brief pause and then a resumption without operator involvement.

Practice on PipeCode

• Drill the streaming practice library → for the change-data-capture, resume-token, and event-routing problems senior interviewers use to separate CDC muscle memory from theory.
• Rehearse on the ETL practice library → for the document-store ingest, hydration-cost, and per-sink projection patterns that turn a raw change stream into a downstream-ready pipeline.
• Sharpen the tuning axis with the optimization practice library → for the oplog-window, updateLookup cost, and shard-lag problems that show up on every senior Mongo CDC scorecard.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the change-stream + resume-token intuition against real graded inputs.

Lock in MongoDB change-stream muscle memory

Docs explain the API. PipeCode drills explain the decision — when to enable `updateLookup`, when Debezium beats the native driver, how to size the oplog so your consumer survives a real outage. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice streaming problems →
Practice ETL problems →