[Kafka-connect] Streaming the data of MongoDB through Kafka-connect into topic Kafka.

1) Install MongoDB

Nếu bạn chưa rành về mongoDB có thể sem vài chiêu thức cơ bản ở đây:

Reference links:
https://gist.github.com/asoorm/7822cc742831639c93affd734e97ce4f

Mình cung cấp link docker-compose.yaml như sau:

version: '3.3'
services:
  mongo-db:
    image: mongo:4.0.5
    hostname: mongo-db
    container_name: mongo-db
    expose:
      - "27017"
    ports:
      - "27017:27017"
    entrypoint: [ "/usr/bin/mongod", "--replSet", "rs0", "--bind_ip_all" ]
    volumes:
      - ./mongo-db:/data

Giờ bạn run:
docker-compose up -d

thực hiện hiện exec vào container và tạo database, collection và insert data.

docker exec -it mongo-db bash

mongo

Hiện tại thì chúng ta chạy mongodb ở mode replicaSet thì kafka-connect với work được vụ CDC source nhé. Bạn để ý mình có run 1 command trong mongoDB
entrypoint: [ "/usr/bin/mongod", "--bind_ip_all", "--replSet", "rs0" ]

Lúc này show database luôn sẽ lỗi bạn tham khoả link dưới nhé:
https://gist.github.com/asoorm/7822cc742831639c93affd734e97ce4f

Bạn run command này

rs.initiate({
      _id: "rs0",
      version: 1,
      members: [
         { _id: 0, host : "<IP_HOST>:27017" }
      ]
   }
)

Tạo DB:

use todo

giờ tạo collection và insert data

db.newtodo.insertOne({"name": "Nimtechnology", age:2, past_clubs: ["Istio", "Kafka-connect", "K8s"], "isRetired": false})

Như bạn thấy ở trên collection sẽ có tên là newtodo và database có tên là todo

2) Create Mongo Source Connector on Kafka-connect

giờ bạn vào lenses tạo connector MongoSourceConnector
https://www.confluent.io/hub/mongodb/kafka-connect-mongodb

Bạn sẽ cần build docker kafka-connect như link dưới:
add plugin via build image and base on cp-kafka-connect-base

connector.class=com.mongodb.kafka.connect.MongoSourceConnector
tasks.max=1
batch.size=0
change.stream.full.document=updateLookup
collection=newtodo
database=todo
topic.prefix=mongouat
poll.await.time.ms=5000
connection.uri=mongodb://192.168.101.27:27017/?replicaSet=rs0
name=source.mongo_connector.todo.v1
value.converter=org.apache.kafka.connect.storage.StringConverter
key.converter=org.apache.kafka.connect.storage.StringConverter
poll.max.batch.size=1000

Nếu mongodb của bạn có user pass thì:

>>>>>>>>>>
connection.uri: mongodb://<username>:<password>@mongodb-0.svr.services:27017,mongodb-1.svr.services:27017,mongodb-2.svr.services:27017/?replicaSet=<name_replicaset>&authSource=admin
>>>>>>>>>

Và bạn cũng cần chú ký phải có các time số sau trong url nhé:
?replicaSet=&authSource=admin

Cái này nó sẽ ko tự tạo topic âu he, Chúng ta phải tạo tay.

https://docs.mongodb.com/kafka-connector/current/source-connector/usage-examples/topic-naming/#topic-prefix-example
bạn đọc cái này để hiểu tạo topic có name là j

Nếu bạn chưa hiểu thì tạo topic với name như sau:
<topic.prefix>.<database>.<collection>