MongoDB Change Streams

Availability

Change streams are available for replica sets and sharded clusters:

• Storage Engine.

  The replica sets and sharded clusters must use the WiredTiger storage engine. Change streams can also be used on deployments that use the encryption-at-rest feature.

• Replica Set Protocol Version.

  The replica sets and sharded clusters must use replica set protocol version 1 (pv1).

• Read Concern "majority" Enablement.

  Change streams are available regardless of the "majority" read concern support; that is, read concern majority support can be either enabled (default) or disabled to use change streams.

Connect

Connections for a change stream can either use DNS seed lists with the +srv connection option or by listing the servers individually in the connection string.

If the driver loses the connection to a change stream or the connection goes down, it attempts to reestablish a connection to the change stream through another node in the cluster that has a matching read preference. If the driver cannot find a node with the correct read preference, it throws an exception.

For more information, see Connection String URI Format.

Watch a Collection, Database, or Deployment

You can open change streams against:

Change Stream Performance Considerations

If the amount of active change streams opened against a database exceeds the connection pool size, you may experience notification latency. Each change stream holds a connection open with a getMore operation while waiting for the next event. To avoid latency, ensure that the pool size is greater than the number of open change streams. For details, see the maxPoolSize setting.

Open a Change Stream

To open a change stream:

• For a replica set, open the change stream from any data-bearing member.

• For a sharded cluster, open the change stream from the mongos.

The following example opens a change stream for a collection and iterates over the cursor to retrieve the change stream documents. [1]

Iterate the change stream cursor to retrieve change events. See Change Stream Events for the change stream document format.

The change stream cursor remains open until one of the following occurs:

• The cursor is explicitly closed.

• An invalidate event occurs; for example, a collection drop or rename.

• The connection to the MongoDB deployment closes or times out. See Behavior for more information.

• If the deployment is a sharded cluster, a shard removal may cause an open change stream cursor to close. The closed change stream cursor may not be fully resumable.

Modify Change Stream Output

Lookup Full Document for Update Operations

By default, change streams only return the delta of fields during the update operation. However, you can configure the change stream to return the most current majority-committed version of the updated document.

The updateLookup operation reads the document identified by its shard key and document identifier from the collection. The collection is identified by its name and uses the collection data as it exists at the time the change stream is processed. Consider these scenarios:

• If the collection is renamed, no document is returned.

• If the collection is renamed and a new collection is created with the old name, then the lookup operation is performed on the new collection. If a matching document is found, it is returned.

Resume a Change Stream

Change streams are resumable by specifying a resume token to either resumeAfter or startAfter when opening the cursor.

snippet install resumetoken
decodeResumeToken('<RESUME TOKEN>')

npx mongodb-resumetoken-decoder <RESUME TOKEN>

{
   "_id": {
      "_data": "82635019A0000000012B042C0100296E5A1004AB1154ACACD849A48C61756D70D3B21F463C6F7065726174696F6E54797065003C696E736572740046646F63756D656E744B65790046645F69640064635019A078BE67426D7CF4D2000004"
    },
    "operationType": "insert",
    "clusterTime": Timestamp({ "t": 1666193824, "i": 1 }),
    "collectionUUID": new UUID("ab1154ac-acd8-49a4-8c61-756d70d3b21f"),
    "wallTime": ISODate("2022-10-19T15:37:04.604Z"),
    "fullDocument": {
       "_id": ObjectId("635019a078be67426d7cf4d2"'),
       "name": "Giovanni Verga"
    },
    "ns": {
       "db": "test",
       "coll": "names"
    },
    "documentKey": {
       "_id": ObjectId("635019a078be67426d7cf4d2")
    }
}

{
   "cursor": {
      "firstBatch": [],
      "postBatchResumeToken": {
         "_data": "8263515EAC000000022B0429296E1404"
      },
      "id": Long("4309380460777152828"),
      "ns": "test.names"
   },
   "ok": 1,
   "$clusterTime": {
      "clusterTime": Timestamp({ "t": 1666277036, "i": 1 }),
      "signature": {
         "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
         "keyId": Long("0")
      }
   },
   "operationTime": Timestamp({ "t": 1666277036, "i": 1 })
}

{
   "cursor": {
      "nextBatch": [],
      "postBatchResumeToken": {
         "_data": "8263515979000000022B0429296E1404"
      },
      "id": Long("7049907285270685005"),
      "ns": "test.names"
   },
   "ok": 1,
   "$clusterTime": {
      "clusterTime": Timestamp( { "t": 1666275705, "i": 1 } ),
      "signature": {
         "hash": Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
         "keyId": Long("0")
      }
   },
   "operationTime": Timestamp({ "t": 1666275705, "i": 1 })
}

Use Cases

Change streams can benefit architectures with reliant business systems, informing downstream systems once data changes are durable. For example, change streams can save time for developers when implementing Extract, Transform, and Load (ETL) services, cross-platform synchronization, collaboration functionality, and notification services.

Access Control

For deployments enforcing Authentication on Self-Managed Deployments and authorization:

• To open a change stream against specific collection, applications must have privileges that grant changeStream and find actions on the corresponding collection.

  { resource: { db: <dbname>, collection: <collection> }, actions: [ "find", "changeStream" ] }

• To open a change stream on a single database, applications must have privileges that grant changeStream and find actions on all non-system collections in the database.

  { resource: { db: <dbname>, collection: "" }, actions: [ "find", "changeStream" ] }

• To open a change stream on an entire deployment, applications must have privileges that grant changeStream and find actions on all non-system collections for all databases in the deployment.

  { resource: { db: "", collection: "" }, actions: [ "find", "changeStream" ] }

Event Notification

Change streams only notify on data changes that have persisted to a majority of data-bearing members in the replica set. Change streams ensure that notifications trigger only for majority-committed changes that are durable in failure scenarios.

For example, consider a 3-member replica set with a change stream cursor opened against the primary. If a client issues an insert operation, the change stream only notifies the application of the data change once that insert has persisted to a majority of data-bearing members.

If an operation is associated with a transaction, the change event document includes the txnNumber and the lsid.

Collation

Change streams use simple binary comparisons unless an explicit collation is provided.

Change Streams and Orphan Documents

Starting in MongoDB 5.3, during range migration, change stream events are not generated for updates to orphaned documents.

Change Streams with Document Pre- and Post-Images

Starting in MongoDB 6.0, you can use change stream events to output the version of a document before and after changes (the document pre- and post-images):

• The pre-image is the document before it was replaced, updated, or deleted. There is no pre-image for an inserted document.

• The post-image is the document after it was inserted, replaced, or updated. There is no post-image for a deleted document.

• Enable changeStreamPreAndPostImages for a collection using db.createCollection(), create, or collMod. For example, when using the collMod command:

  db.runCommand( {
     collMod: <collection>,
     changeStreamPreAndPostImages: { enabled: true }
  } )

Pre- and post-images are not available for a change stream event if the images were:

• Not enabled on the collection at the time of a document update or delete operation.

• Removed after the pre- and post-image retention time set in expireAfterSeconds.

  • The following example sets expireAfterSeconds to 100 seconds on an entire cluster:

    use admin
    db.runCommand( {
       setClusterParameter:
          { changeStreamOptions: {
             preAndPostImages: { expireAfterSeconds: 100 }
          } }
    } )

    Note

    The setClusterParameter command is not supported in MongoDB Atlas clusters. For information on Atlas support for all commands, see Unsupported Commands in Atlas.

  • The following example returns the current changeStreamOptions settings, including expireAfterSeconds:

    db.adminCommand( { getClusterParameter: "changeStreamOptions" } )

  • Setting expireAfterSeconds to off uses the default retention policy: pre- and post-images are retained until the corresponding change stream events are removed from the oplog.

  • If a change stream event is removed from the oplog, then the corresponding pre- and post-images are also deleted regardless of the expireAfterSeconds pre- and post-image retention time.

Additional considerations:

• Enabling pre- and post-images consumes storage space and adds processing time. Only enable pre- and post-images if you need them.

• Limit the change stream event size to less than 16 mebibytes. To limit the event size, you can:

  • Limit the document size to 8 megabytes. You can request pre- and post-images simultaneously in the change stream output if other change stream event fields like updateDescription are not large.

  • Request only post-images in the change stream output for documents up to 16 mebibytes if other change stream event fields like updateDescription are not large.

  • Request only pre-images in the change stream output for documents up to 16 mebibytes if:

    • document updates affect only a small fraction of the document structure or content, and

    • do not cause a replace change event. A replace event always includes the post-image.

• To request a pre-image, you set fullDocumentBeforeChange to required or whenAvailable in db.collection.watch(). To request a post-image, you set fullDocument using the same method.

• Pre-images are written to the config.system.preimages collection.

  • The config.system.preimages collection may become large. To limit the collection size, you can set expireAfterSeconds time for the pre-images as shown earlier.

  • To monitor the size of config.system.preimages, connect to a shard node on a sharded cluster or a mongod node on a replica set. Then, run the following commands:

    use config
    db.system.preimages.totalSize()
    db.system.preimages.stats()

    Note

    You must have the collStats privilege action on the config.system.preimages collection to run these commands.

    To run these commands on an MongoDB Atlas deployment, you must have the atlasAdmin role.

  • Pre-images are removed asynchronously by a background process.

For complete examples with the change stream output, see Change Streams with Document Pre- and Post-Images.

If the initialSyncMethod parameter for the cluster is fileCopyBased, then there is no impact on change stream listeners.

If initialSyncMethod is logical and a change stream is opened on a newly synchronized node and reads events from a point in time earlier than the completion of the logical initial sync, the pre- and post-images may be missing.