Kafka

2. Kafka - Connector

Seonho Jeong

04 5월 2023 • 4 min read

MongoDB replication & writeConcern

Debezium, Mongodb Kafka Connector 등 MongoDB CDC 를 붙이려면 우선적으로 이해해야 하는 부분이 MongoDB oplog고 oplog를 이해하려면 MongoDB Replication과 writeConcern에 대한 이해가 필요하다.
replication 환경이나 sharding 환경에서 oplog를 사용할 수 있으며, write 관련된 처리는 writeConcern 설정에 의해 제어됨

Replication

https://www.mongodb.com/docs/manual/replication/
replica set 은 같은 데이터 셋을 유지관리하는 mongod (mongodb daemon set)의 집합
Replica Set은.. 뭐 고가용성 제공, 리얼 배포 환경의 기본 (리얼 서비스에는 기본적으로 레플리카셋을 이용해라)

Redundancy and Data Availability
여러개 서버를 띄우는 것이니 고가용성 제공. readConcern 설정에 따라 읽기 성능도 좋아질 수 있음 (분산 환경이 되다보니)

Replication in MongoDB
앞선 내용은 대부분의 시스템에서 말하는 내용 (복제본을 만드니... 고가용성 + 성능)
MongoDB에서는 어떻게 복제를 하고 신뢰성을 얻을지?

Replica Set은 data bearing 노드들과 optionally one arbiter 노드로 구성

data bearing : 1개의 primary 노드와 나머지 secondary 노드
- primary : 쓰기 관련된 모든 실행을 맡음. 오로지 하나만 primary!
  - 근데 네트워크 지연으로 인해 특정 상황에서는 primary가 2개일 수 있음. 이 경우에는 이전 primary와 새 primary가 같이 존재하는 상태. 이전 primary가 네트워크 지연으로 인해 heartbeat 응답이 느려졌다거나 아무튼.. 이런 경우엔 이전 primary에서 발생한 쓰기는 stale data로 판단하여 롤백한다.
  - primary 노드는 모든 데이터 셋 변경에 대한 로그를 operation log에 저장한다. (i.e. oplog )
- secondary : primary의 oplog를 복제해가고 그 후 oplog를 읽으며 데이터 변경을 수행, primary와 동일한 데이터셋을 가지게 된다.
  - primary가 에러 상황일 때, primary 에 적합한 secondary 노드를 선출한다.
arbiter : secondary 노드 하나, primary 노드 하나만 있고 더이상 노드 추가할 리소스가 없을 때 단순히 primary 선출 투표만하는 노드, primary가 secondary가 되거나, secondary가 primary가 되어도 arbiter는 여전히 arbiter로 유지된다.

primary 투표는 여러 상황에서 일어나는데.

새 노드가 replica set에 추가
레플리카 셋 초기화
설정된 timeout 보다 오랫동안 secondary 멤버들이 primary 노드와 연결이 끊어졌을 때

각 노드들은 2초마다 서로 hearbeat를 전송 연결 유지되는지 확인하고 10초간 응답이 없으면 해당 노드를 연결 끊어진 것으로 판단한다.

투표는 어떻게 진행?
primary 노드가 안정된 후, 투표 알고리즘은 secondary 노드들에게 투표 우선순위를 정해준다. 우선순위는 투표 시기와 결과에 영향을 준다. 높은 우선순위를 가진 노드는 낮은 우선순위를 가진 노드보다 먼저 투표를 실시, 투표 결과에 있어서도 선출될 확률이 더 높다.

writeConcern

https://wonyong-jang.github.io/aws/2021/05/24/AWS-DocumentDB-Read-Write-Option.html
client가 primary에 쓰기 요청하고 응답받는데
mongodb의 replication은 비동기형태라 primary에서 응답한 뒤 secondary에서 data replicate 하기 전 이 때 primary가 에러상황이 되고 data replicate를 하지 못한 secondary가 primary 되면 데이터 손실이 일어난다.

기본적으로는 primary에만 기록 완료되면 client에게 완료 응답을 한다.
w option
client에게 응답 보내기 위해 필요한 최소한의 기록 완료된 멤버의 수.
멤버가 replicate 완료되었다고 응답을 주는데 primary 본인을 포함하여 쓰기 완료된 멤버의 수가 w option 이상이면 client에게 완료 메시지를 보낸다.
w = majority인 경우 멤버의 과반수 이상으로 자동 설정된다.

oplog

primary에서 쓰기 operation이 성공했을 경우(write concern에 의해 쓰기 성공에 대한 기준이 생김) oplog를 기록한다.
secondary는 oplog를 복제해가서 본인의 데이터 셋에서 oplog를 operate해 primary와 데이터를 일치시킨다.
MongoDB CDC를 구현하기 위해 oplog를 사용.

oplog example

_id.data 가 resumeToken
https://www.mongodb.com/docs/manual/changeStreams/#resume-tokens-from-change-events

Insert

 
{
  "_id": {
    "_data": ...
  },
  "operationType": "insert",
  "clusterTime": {
    "$timestamp": "7234699619282714625"
  },
  "fullDocument": {
  ...
  },
  "ns": {
    "db": "test",
    "coll": "Account"
  },
  "documentKey": {
    "_id": ...
  }
}

Update

 
{
  "_id": {
    "_data": ...
  },
  "operationType": "update",
  "clusterTime": {
    "$timestamp": "7234699739541798916"
  },
  "ns": {
    "db": "test",
    "coll": "Account"
  },
  "documentKey": {
    "_id": ...
  },
  "updateDescription": {
    "updatedFields": {
      "fieldA": ...,
      "fieldB": {
        "fieldB-1": ...
      },
      "fieldC": ...
    },
    "removedFields": ["fieldD", "fieldE"]
  }
}

Delete

 
{
  "_id": {
    "_data": ...
  },
  "operationType": "delete",
  "clusterTime": {
    "$timestamp": "7234700035894542337"
  },
  "ns": {
    "db": "test",
    "coll": "Account"
  },
  "documentKey": {
    "_id": ...
  }
}