Kafka CDC: MySQL to MongoDB 동기화 #1

 

[그림 1] Kafka CDC

1. Debezium MySQL Source Connector란?

Debezium은 데이터베이스의 변경 이벤트를 감지하고 이를 Kafka에 발행해주는 CDC 플랫폼인데, MySQL Source Connector는 binlog를 기반으로 작동하여 INSERT, UPDATE, DELETE 이벤트를 실시간으로 처리한다.

2. binlog란?

MySQL의 binlog(binary log)는 데이터 변경 사항을 이진 포맷으로 기록한 로그파일이다. binlog은 주로 MySQL replication에서 사용된다. 쉽게 말하면 binlog는 데이터가 변경되는 모든 이벤트를 가지고 있기때문에 replication 혹은 데이터 복구 등에 사용될 수 있다.

3. Debezium의 동작 원리

1. MySQL에 replication client로 접속
   • Debezium은 MySQL에 replication 권한을 가진 유저로 접속한다. 그 이유는 binlog 데이터를 읽을 수 있는 권한을 확보하기 위해서다.
   • 그렇기 때문에 MySQL Source Connector에서 사용하는 유저는 replication 권한을 부여해주어야한다.
2. binlog 이벤트 읽기 및 해석
   • binlog가 갱신되면 Debezium 커넥터는 이를 지속적으로 읽고, 이진 포맷을 해석하여 어떤 테이블에서 어떤 변경이 발생했는지 파악한다.
3. CDC 이벤트 생성 및 Kafka 발행
   • 해석된 데이터를 JSON 메시지로 변환하여 Kafka Topic에 발행한다.
   • 발행되는 메시지의 예는 아래와 같다.

{
  "op": "u",
  "before": {"id":1, "link":"link-1"},
  "after": {"id":1, "link":"link-x"}
}

4. Debezium이 binlog를 읽는 시점

• Snapshot 단계
  • 초기 구동 시 전체 테이블을 SELECT 하여 Kafka에 전송하고 binlog 위치도 기록한다.
• Binlog Streaming 단계
  • 스냅샷 시점 이후의 binlog를 실시간으로 읽어 이벤트를 처리한다.
• Connector 재시작 시 동작
  • Kafka offset 저장소에 마지막 읽은 binlog 위치를 저장하고, 장애 복구 시 해당 위치부터 스트리밍을 재개한다.

참고로, snapshot 없이 오직 binlog만을 이용하도록 설정할 수도 있다.

 

 

5. binlog 위치 저장과 offset 관리

스냅샷 단계가 끝나면, Debezium MySQL Source Connector는 MySQL 서버에게 현재 binlog 파일명과, position(오프셋)을 요청해서 받는다. 이 정보를 offset 저장소(OFFSET_STORAGE_TOPIC 설정 값)에 저장한다. 그렇기때문에 재시작이 되어도 데이터 중복 처리가 되지 않고 안전하게 스트리밍할 수 있는 것이다.

binlog offset 정보의 예시는 아래와 같다.

{
  "ts_sec": 1751609233,
  "file": "mariadb-bin.000001",
  "pos": 11899,
  "row": 1,
  "server_id": 1
}

6. 커넥터 설정 및 config 수정 방법

운영 중 소스 테이블이 변경될 경우, config 수정을 통해 반영할 수 있다

curl -v -X PUT http://localhost:8083/connectors/my-source-connector/config \
-H "Content-Type: application/json" \
-d @source-connector/cdc-source-connector.json

7. 운영 중 발생하는 고려사항

앞서 이야기한 것 처럼 Debezium MySQL Source Connector은 스냅샷 단계에서 binlog를 읽고 offset을 기록한다고했다. 눈치가 빠르다면 이 부분에서 알 수 있는 것이있다.

• 새로운 테이블을 추가할 경우
  • 해당 테이블에 대한 스냅샷을 미리 수행하지 않으면 기존 데이터는 Kafka에 발행되지 않는다.
  • 그렇기 때문에 반드시 초기 동기화를 고려해야 한다.
• Kafka Connect가 shutdown되었을 경우
  • 어떠한 이슈로 Kafka Connect가 내려갔다면 대응 후 다시 가동할텐데, 이때 Kafka offset 저장소로부터 binlog를 어디까지 읽었는지 알 수 있기 때문에 Kafka Connect가 재가동 되면 다시 offset부터 읽어 스트리밍할 수 있다.
• Kafka cluster가 shutdown되었을 경우
  • 브로커가 하나 이상 shutdown된다면 DisconnectException이 발생되면서 kafka connect가 주기적으로 계속 연결을 시도한다.
  • Kafka cluster가 다시 복구가 되면 리밸런싱이 마치고나서 kafka connect에서 저장된 offset부터 다시 binlog를 읽기때문에 스트리밍이 되지 않은 부분부터 최신까지 다시 스트리밍한다.
  • 만약 데이터 용량이 크다면 lag이 높아질 수 있다는 점에 유의해야한다.
  • Kafka cluster가 다시 복구가 되면 당연히 리밸런싱이 진행되므로 리밸런싱 중 source connector를 재시작하면 아래와 같은 응답을 받을 수 있다.

curl -X POST http://localhost:8083/connectors/cdc-mysql-source-connector/restart {"error_code":409,"message":"Cannot complete request momentarily due to no known leader URL, likely because a rebalance was underway."}

8. References

• Debezium MySQL Soure Connector 공식 문서
• Debezium 아키텍처
• Kafka 공식 문서
• Kafka connect rest api