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

Contents

1) Install MongoDB
2) Create Mongo Source Connector on Kafka-connect

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:

[MongoDB] Install and config MongoDB is so easy on Docker.

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" }
      ]
   }
)

Lưu ý: bạn phải run command trên xong thì mới kêt nối kiểu Replica set được!

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

Nếu bạn thấy response như thế này là data đã được insert thành công.

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>