[kafka] mongodb-source-connector 사용해서 mongodb 변경 값 실시간으로 메시지 publish

1. kafka Tutorial 실습

공식 Docs : https://www.mongodb.com/docs/kafka-connector/current/tutorials/source-connector/

git clone https://github.com/mongodb-university/kafka-edu.git

* json파일로 connector 에 관한 내용을 세팅
1. cx ~.json 으로 connector 적용
2. del <connector name> 으로 connector 삭제

# simplesource.json 파일 예시
{
  "name": "mongo-simple-source",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "connection.uri": "mongodb://mongo1",
    "database": "Tutorial1",
    "collection": "orders"
  }
}

---

2. Docker container 의 kafka 와 로컬 컴퓨터의 MongoDB 연결

change stream 방법은 replica 생성되어있는 mongodb에서만 사용할 수 있음!

# mongod.conf 파일에 replica 설정!


replication:
  oplogSizeMB: 2000
  replSetName: "rs0"

---

1. 실습 docker-compose 파일의 mongodb 부분은 주석처리 후 docker-compose up
   • source.json 파일을 volume 처리하여 connect container 내부에서 사용할 수 있게 함.

{
  "name": "mongo-simple-source",
  "config": {
    "connector.class": "com.mongodb.kafka.connect.MongoSourceConnector",
    "connection.uri": "mongodb://<로컬 IP>:27017",
    "database": "Tutorial1", # 변경 사항 확인할 DB
    "collection": "orders" # 변경 사항 확인할 collection
  }
}

---

2. connect container 접속
   • / (root) 경로
   • /bin 폴더 내에 sh 파일 존재

---

3. REST API로 connector 생성 & 조회 & 삭제

# 조회
curl -X GET "http://localhost:8083/connectors"

# 생성
curl -X POST -H "Content-Type application/json" -d @<connector.json 파일명> http://localhost:8083/connectors

# 삭제
curl -X DELETE "http://localhost:8083/

---

4. Topic 조회용 sh 실행

/bin/kafka-console-consumer --bootstrap-server broker:29092 --topic <DB명>.<collection명> --from-beginning

* --from-beginning 옵션 : Topic에 쌓인 모든 메시지를 처음부터 로드
* mongodb-kafka-connector에서 <db.coll> 은 topic명으로 사용될 수 있다.

5. DB data 변경하고 결과 확인 ^___^ v

 *DB 수정되자마자 Topic에 쌓이는 것을 확인할 수 있다.

---

+ Docker에서 외부 로컬 호스트 연결이 안 될 때 참고

* docker container 내부에서 localhost:port를 쓴다면, 이때의 localhost는 외부 로컬 호스트(docker 실행한 로컬 컴퓨터)가 아닌 docker 내부의 로컬호스트를 의미.
* 우분투에서는 로컬 컴퓨터의 외부 IP로 넣어주기, 윈도우와 mac에서는 host.docker.internal 을 ip로 사용하면 된다고 함!
+ 그래도 안된다면 로컬 컴퓨터 mongod.conf 파일 확인.

net:
port: 27017
bindIp: 0.0.0.0  --> 이 부분이 0.0.0.0 맞는지 확인! (외부 접근 허용)