Using MongoDB Change Streams for Real‑Time Event Propagation in Microservices

Problem

Microservices often need to react to data changes in a source database without polling or custom triggers. Polling adds latency, increases load, and complicates consistency guarantees. Custom triggers require database‑side logic that can be brittle and hard to maintain across schema changes.

Takeaway

MongoDB Change Streams provide a low‑latency, server‑side notification mechanism that can be wired into a message broker (e.g., Kafka). A lightweight consumer service can translate collection changes into domain events, preserving idempotency and auditability.

Requirements

• MongoDB 4.4+ running as a replica set or sharded cluster (standalone does not support change streams).
• Application role with changeStream and find privileges on the target database.
• Persistent storage for resume tokens (e.g., a separate collection or key‑value store).
• Message broker or event bus to forward events to downstream services.
• Monitoring for oplog tailing lag and consumer health.

Minimal Suitable Design

The simplest architecture is a dedicated consumer service that:

1. Opens a change‑stream cursor on the collection’s namespace.
2. Validates each event against a JSON schema.
3. Publishes the event to a broker.
4. Persists the resume token after each successful publish.

Diagram labels: Replica Set → Change‑Stream Consumer → Message Broker → Downstream Microservice.

Node.js Consumer Skeleton

const { MongoClient } = require("mongodb");
const kafka = require("kafka-node");

const uri = "mongodb://user:pass@replica1:27017,replica2:27017/?replicaSet=rs0";
const client = new MongoClient(uri, { useUnifiedTopology: true });
const dbName = "orders";
const collectionName = "orders";

async function run() {
  await client.connect();
  const coll = client.db(dbName).collection(collectionName);

  // Load last resume token
  const meta = await client.db("meta").collection("resumeTokens").findOne({
    collection: collectionName,
  });
  const options = meta ? { resumeAfter: meta.token } : { startAfter: null };

  const cursor = coll.watch([], options);
  const producer = new kafka.KafkaClient({ kafkaHost: "broker:9092" });
  const kafkaProducer = new kafka.Producer(producer);

  cursor.on("change", async (change) => {
    try {
      // Simple schema validation
      if (!change.fullDocument || !change.fullDocument._id) return;

      const event = {
        type: change.operationType,
        payload: change.fullDocument,
        metadata: {
          ns: change.ns,
          ts: change.clusterTime,
        },
      };

      // Publish to Kafka
      kafkaProducer.send(
        [{ topic: "orders-events", messages: JSON.stringify(event) }],
        (err) => {
          if (err) throw err;
        }
      );

      // Persist resume token
      await client
        .db("meta")
        .collection("resumeTokens")
        .updateOne(
          { collection: collectionName },
          { $set: { token: change._id } },
          { upsert: true }
        );
    } catch (e) {
      console.error("Event processing failed", e);
      // Decide on retry policy
    }
  });
}

run().catch(console.error);

Trust & Data Boundaries

• Event Validation: Schema validation prevents malformed documents from being propagated.
• Idempotency: Store the _id of the last processed event. Use it to guard against duplicate processing on restarts.
• Resume Token Security: Keep the token in a protected collection with limited access. Do not expose it via APIs.
• Access Control: The consumer’s MongoDB role should be minimal – only changeStream on the target database.

Operational Checks

Check	Command / Metric	Target
Oplog size & tail lag	rs.printReplicationInfo() + system.profile	≤ 5 s
Consumer health	Prometheus metrics (e.g., change_stream_lag_seconds)	≤ 1 s
Resume token persistence	Query meta.resumeTokens for recent token	Token matches last _id

Failure Modes & Mitigations

• Oplog exhaustion: High write volume can fill the oplog faster than the consumer processes events. Mitigate by increasing oplog size or scaling consumer instances.
• Missing events due to resume token loss: Persist the token to durable storage and verify on startup. Use startAfter only if you can guarantee no missed events.
• Duplicate events: Implement idempotent handlers on downstream services. Use the event’s _id as a deduplication key.
• Cluster reconfiguration: Replica set changes (e.g., adding members) do not break change streams but may increase lag. Monitor for changes via rs.status().
• Security breach: Leaked resume tokens can replay events. Enforce strict ACLs and rotate tokens periodically.

When to Change the Design

• If the write load exceeds the oplog’s capacity, switch to a multi‑consumer architecture or shard the collection.
• When operating on MongoDB Atlas free tier, upgrade to a paid tier or deploy a self‑hosted replica set.
• If downstream services require strict ordering across multiple collections, consider a dedicated ordering layer or partitioned Kafka topics.
• When schema evolution becomes frequent, integrate a schema registry to manage event contracts.

Example Verification Steps

1. Deploy a 3‑node replica set:

   mongod --replSet rs0 --port 27017 --dbpath /data/db1
   mongod --replSet rs0 --port 27018 --dbpath /data/db2
   mongod --replSet rs0 --port 27019 --dbpath /data/db3
   
   mongo --port 27017
   rs.initiate({ _id: "rs0", members: [ { _id: 0, host: "localhost:27017" }, { _id: 1, host: "localhost:27018" }, { _id: 2, host: "localhost:27019" } ]})
   

2. Enable change stream in the shell:

   use orders
   const coll = db.orders;
   const stream = coll.watch();
   stream.forEach(doc => printjson(doc));
   

3. Insert test documents and observe events in the consumer logs.
4. Persist resume token to meta.resumeTokens and restart the consumer. Verify no duplicate events are emitted.
5. Run a burst write test (e.g., 10k inserts per second) and monitor change_stream_lag_seconds via Prometheus.

Conclusion

MongoDB Change Streams offer a robust, low‑latency path from database mutations to event‑driven architectures. By isolating the consumer, persisting resume tokens, and monitoring oplog lag, teams can scale event propagation without polling or custom triggers. The design remains lightweight yet extensible, and can be adapted for high‑write workloads or stricter ordering requirements by adding consumer instances or integrating a schema registry.