cdc-connector v1.2.7

CDC Connector

The CDC Connector is a tool designed to capture and stream changes from MongoDB collections real-time.

Features

• Real-time change data capture
• Support for multiple MongoDB change streams
• Support for transactional outbox pattern
• Deliver changes and messages directly to NATS
• High performance and reliability

Getting Started

• To start using the CDC Connector, clone the repository first:

git clone git@github.com:Ledn-io/cdc-connector.git
cd cdc-connector

• To run the CDC Connector, create a .env file first and then execute:

cp .env.sample .env
nano .env
npm run docker:start

Configuration

Adding a new capture collection

• Add in config/ledn-api-collections-array.ts the collection name
• Add in config/capture-config/get-capture-config.ts a new map with collection-name -> collectionName-capture-config.ts class instance.

• Add in config/stream-config/get-stream-config.ts a new map with collection-name -> collectionName-stream-config.ts class instance.

• Add in config/capture-config/collections folder a file with collectionName-capture-config.ts. Choose a subject function that fits your use case, or create a new one.

{
  "database": "my-db",
  "collection": "my-collection",
  "stream": "cdc-collection-stream",
  "subject": getSubjectForDocument
}

• Add in config/stream-config/collections folder a file with collectionName-stream-config.ts. Choose the correct values for your use case, with the following schema:

{
  "name": "cdc-collection-stream",
  "subjects": ["cdc.connector.my-db.my-collection.>"],
  "retention": "limits",
  "max_consumers": -1,
  "max_msgs_per_subject": 1,
  "max_msgs": -1,
  "max_bytes": -1,
  "max_age": 0,
  "max_msg_size": -1,
  "storage": "file",
  "discard": "new",
  "num_replicas": 3,
  "duplicate_window": 120000000000,
  "sealed": false,
  "deny_delete": false,
  "deny_purge": false,
  "allow_rollup_hdrs": false,
  "allow_direct": true,
  "mirror_direct": false,
  "discard_new_per_subject": true
}

For an explanation of each configuration please visit the official documentation for NATS streams.

• Add new nats permissions for each new collection, eg. see nats/server.conf -> $JS.API.STREAM.CREATE.cdc-configs-stream

  • Add the new necessary permissions in the cdc group in the synadia interface inside the group permissions for the user in the application, eg. in dev is cdc-connector-dev -> group cdc-connector-dev-sk, edit and then go to permissions and click add, copy your new permission, eg. in dev is cdc-connector-dev in the synadia interface.

• Add permission for the cdc-user in each environment database for the new collections,

  • Create a ticket in #ask-devops channel

Adding a new outbox

WARNING: DO NOT add multiple outbox configurations for the same outbox collection. If you are just adding support for a new Aggregate Type, just add a new route on an existing outbox.

Add a new outbox config file in the config/outboxes folder with the following schema:

{
  "database": "my-db",
  "collection": "my-outbox-collection",
  "resumeTokensStore": "cdc-resume-tokens",
  "routes": [
    {
        "aggregateType": "MyAggregate",
        "subject": "cdc.connector.outbox.my-db.my-aggregate",
        "stream": "my-outbox-stream"
    }
  ]
}

Add a new stream config file in the config/streams folder with the following schema:

{
  "name": "my-outbox-stream",
  "subjects": ["cdc.connector.outbox.my-db.my-aggregate.>"],
  "retention": "limits",
  "max_consumers": -1,
  "max_msgs_per_subject": 1,
  "max_msgs": -1,
  "max_bytes": -1,
  "max_age": 0,
  "max_msg_size": -1,
  "storage": "file",
  "discard": "new",
  "num_replicas": 3,
  "duplicate_window": 120000000000,
  "sealed": false,
  "deny_delete": false,
  "deny_purge": false,
  "allow_rollup_hdrs": false,
  "allow_direct": true,
  "mirror_direct": false,
  "discard_new_per_subject": true
}