Deploying CDP Mongo Sink Connectors

This section provides detailed instructions on how to deploy HCL CDP Mongo Sink Connectors via the Kafka Connect REST API.

Procedure

Identify the Mongo Sink Connector pod in the target namespace.
```
kubectl get pods -n <namespace> | grep mongo-sink-connector
```

Open a shell session in the Mongo Sink Connector pod.

kubectl exec -it cdp-mongo-sink-connector-bf4bd5ddf-qvlnx -- sh -n <namespace>

Run the curl requests to create the Mongo Sink connectors. Replace the connection.uri value with the environment-specific MongoDB connection URI.

Create the kafka-segdb-connect-sst connector:

curl --location 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "kafka-segdb-connect-sst",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
    "errors.log.include.messages": "true",
    "mongo.errors.log.enable": "true",
    "transforms.tsConverter.field": "expts",
    "writemodel.strategy": "com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy",
    "tasks.max": "12",
    "namespace.mapper.value.collection.field": "campaignId",
    "transforms": "customTransformer,tsConverter",
    "transforms.tsConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
    "errors.deadletterqueue.context.headers.enable": "true",
    "mongo.errors.tolerance": "all",
    "database": "segmentation",
    "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
    "namespace.mapper.error.if.invalid": "false",
    "transforms.tsConverter.target.type": "Date",
    "document.id.strategy.partial.value.projection.type": "AllowList",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "document.id.strategy.overwrite.existing": "true",
    "topics": "dmp_sst_profile",
    "collection": "profile",
    "namespace.mapper": "com.mongodb.kafka.connect.sink.namespace.mapping.FieldPathNamespaceMapper",
    "transforms.customTransformer.type": "com.hcl.cdp.kafka.connect.transforms.MongoSinkTransformer$Value",
    "document.id.strategy.partial.value.projection.list": "key_type,key",
    "errors.deadletterqueue.topic.name": "mongo_profile_upsert_dlq",
    "connection.uri":"",
    "name": "kafka-segdb-connect-sst",
    "value.converter.schemas.enable": "false",
    "errors.tolerance": "all",
    "post.processor.chain": "com.mongodb.kafka.connect.sink.processor.DocumentIdAdder",
    "transforms.tsConverter.format": "yyyy-MM-dd"
  }
}'

curl --location 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "profile-data-sync-6550",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "publish.full.document.only": "true",
    "output.format.value": "json",
    "tasks.max": "1",
    "change.stream.full.document": "updateLookup",
    "copy.existing.pipeline": "[]",
    "rotate.interval.ms": "300000",
    "collection": "VIZVRM6550",
    "pipeline": "[]",
    "database": "segmentation",
    "topic.namespace.map": "{\"segmentation.VIZVRM6550\":\"profile-data-sync-6550\"}",
    "connection.uri": "",
    "value.converter.schemas.enable": "false",
    "name": "profile-data-sync-6550",
    "copy.existing": "true",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "poll.max.batch.size": "1"
  }
}'

curl --location 'http://localhost:8083/connectors' \
--header 'Content-Type: application/json' \
--data-raw '{
  "name": "mongo-data-sync-6550",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "publish.full.document.only": "true",
    "output.format.value": "json",
    "tasks.max": "1",
    "change.stream.full.document": "updateLookup",
    "copy.existing.pipeline": "[{\"$match\": {\"masterid\": {\"$ne\": null}}}]",
    "collection": "VIZVRM6550",
    "pipeline": "[{\"$match\": {\"fullDocument.masterid\": {\"$ne\": null}}}]",
    "database": "segmentation",
    "topic.namespace.map": "{\"segmentation.VIZVRM6550\":\"mongo-data-sync-6550\"}",
    "connection.uri": "",
    "value.converter.schemas.enable": "false",
    "name": "mongo-data-sync-6550",
    "copy.existing": "true",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "poll.max.batch.size": "100"
  }
}'

Verify the connector status.

curl http://localhost:8083/connectors/mongo-data-sync-6550/status
curl http://localhost:8083/connectors/profile-data-sync-6550/status
curl http://localhost:8083/connectors/kafka-segdb-connect-sst/status