DOCSP-15809 custom pipeline (#155)

* DOCSP-15809: Usage Example Custom Pipeline

Author: Chris Cho

source/source-connector/fundamentals/change-streams.txt

@@ -1,3 +1,5 @@
+.. _kafka-source-change-streams:
+
 ==============
 Change Streams
 ==============
@@ -14,14 +16,14 @@ Overview
 --------
 
 In this guide, you can learn about **change streams** and how they are
-used in a {+mkc+} source connector. 
+used in a {+mkc+} source connector.
 
 Change Streams
 --------------
 
 Change streams are a feature of MongoDB that allow you to receive real-time
 updates on data changes. Change streams return
-**change event documents**. A change event document is a document in your 
+**change event documents**. A change event document is a document in your
 **oplog** that contains idempotent instructions to recreate a change that
 occurred in your MongoDB deployment as well as the metadata related to that
 change.
@@ -36,24 +38,35 @@ works.
    Change streams are available for replica sets and sharded clusters. A
    standalone MongoDB instance cannot produce a change stream.
 
-For more information on change streams, see the following MongoDB manual entries:
+For more information on the oplog, see the MongoDB manual entry on the
+:manual:`Replica Set Oplog </core/replica-set-oplog/>`.
+
+You can use an aggregation pipeline to modify change stream events in one or
+more of the following ways:
+
+- Filter change events by operation type
+- Project specific fields
+- Update the value of fields
+- Add additional fields
+- Trim the amount of data generated by the change stream
 
-- :manual:`Change streams </changeStreams/>`
-- :manual:`Oplog </core/replica-set-oplog/>`
+For a list of aggregation operators you can use with a change stream, see
+the guide on :manual:`Modify Change Stream Output <changeStreams/#modify-change-stream-output>`
+in the MongoDB manual.
 
 Change Event Structure
 ~~~~~~~~~~~~~~~~~~~~~~
 
 You can find the complete structure of change event documents, including
-descriptions of all fields, 
-:ref:`in the MongoDB manual <change-stream-output>`. 
+descriptions of all fields,
+:ref:`in the MongoDB manual <change-stream-output>`.
 
 .. note:: The Full Document Option
 
    If you want Kafka Connect to receive just the document created or modified
    from your change operation, use the ``publish.full.document.only=true``
-   option. For more information see our
-   guide on <TODO: Link to source connector properties>.
+   option. For more information, see the :ref:`<source-configuration-change-stream>`
+   page.
 
 Source Connectors
 -----------------
@@ -67,7 +80,7 @@ connector closes its change stream when you stop it.
 
    Your source connector uses the MongoDB Java driver to create a change
    stream. For more information, see this guide on change streams in the MongoDB
-   java driver. <TODO: Link to Watch For Changes Usage Example>
+   Java driver. <TODO: Link to Watch For Changes Usage Example>
 
 Resume Tokens
 ~~~~~~~~~~~~~
@@ -76,7 +89,7 @@ Your connector stores a **resume token** to keep track of what changes
 it has processed. A resume token is a piece of data that references
 the ``_id`` field of a change event document in your MongoDB oplog.
 Your connector only processes relevant change event documents written to the oplog after the
-document referenced by its resume token. 
+document referenced by its resume token.
 
 If your source connector does not have a resume token, such as when you start
 the connector for the first time, your connector processes relevant change
@@ -89,5 +102,5 @@ events written to the oplog after it first connects to MongoDB.
 
 If your source connector's resume token does not correspond to any entry in your
 oplog, your connector has an invalid resume token. To learn how to recover from an
-invalid resume token, 
+invalid resume token,
 :ref:`see our troubleshooting guide <kafka-troubleshoot-recover-invalid-resume-token>`.

source/source-connector/usage-examples/custom-pipeline.txt

@@ -1,5 +1,85 @@
-===============
-Custom Pipeline
-===============
+.. _source-usage-example-custom-pipeline:
 
-TODO:
+============================================
+Customize a Pipeline to Filter Change Events
+============================================
+
+.. default-domain:: mongodb
+
+This usage example demonstrates how to configure a **pipeline** to
+customize the data that your source connector consumes. A pipeline is a
+MongoDB aggregation pipeline composed of instructions to the database to
+filter or transform data.
+
+MongoDB notifies the connector of data changes that match your aggregation
+pipeline on a **change stream**. A change stream is a sequence of events that
+describe data changes a client made to a MongoDB deployment in real-time.
+For more information, see the MongoDB Server manual entry on
+:manual:`Change Streams </changeStreams/>`.
+
+
+Example
+-------
+
+Suppose you're an event coordinator needs to collect names and arrival times
+of each guest at a specific event. Whenever a guest checks into the event,
+an application inserts a new document that contains the following details:
+
+.. code-block:: json
+   :copyable: false
+
+   {
+     "_id": ObjectId(...),
+     "eventId": 321,
+     "name": "Dorothy Gale",
+     "arrivalTime": 2021-10-31T20:30:00.245Z
+   }
+
+You can define your connector ``pipeline`` setting to instruct the change
+stream to filter the change event information as follows:
+
+- Create change events for insert operations and omit events for all other
+  types of operations.
+- Create change events only for documents that match the ``fullDocument.eventId``
+  value "321" and omit all other documents.
+- Omit the ``_id`` and ``eventId`` fields from the ``fullDocument`` object
+  using a projection.
+
+To apply these transformations, assign the following aggregation pipeline
+to your ``pipeline`` setting:
+
+.. code-block:: properties
+
+   pipeline=[{"$match": { "$and": [{"operationType": "insert"}, { "fullDocument.eventId": 321 }] } }, {"$project": { "fullDocument._id": 0, "fullDocument.eventId": 0 } } ]
+
+.. important::
+
+   Make sure that the results of the pipeline contain the top-level ``_id``
+   field of the ``payload`` object, which MongoDB uses as the value of the
+   :manual:`resume token </changeStreams/#resume-a-change-stream>`.
+
+When the application inserts the sample document, your configured
+connector publishes the following record to your Kafka topic:
+
+.. code-block:: json
+   :copyable: false
+   :emphasize-lines: 7,8
+
+   {
+     ...
+     "payload": {
+       _id: { _data: ... },
+       "operationType": "insert",
+       "fullDocument": {
+         "name": "Dorothy Gale",
+         "arrivalTime": "2021-10-31T20:30:00.245Z",
+       },
+       "ns": { ... },
+       "documentKey": {
+         _id: {"$oid": ... }
+       }
+     }
+   }
+
+For more information on managing change streams with the source connector, see
+the connector documentation on :ref:`Change Streams <kafka-source-change-streams>`.