[Kafka] Replication (2)

In-Sync Replicas 리스트 관리

• Leader가 관리한다.
• 메시지가 ISR 리스트의 모든 Replica에서 수신되면 Commit된 것으로 간주한다.
• Kafka Cluster의 Controller가 모니터링하는 ZooKeeper의 ISR 리스트에 대한 변경 사항은 Leader가 관리한다.
• Follower가 실패하는 경우, Leader에 의해 ISR 리스트에서 삭제되고  Leader는 새로운 ISR을 사용하여 Commit한다.

1. (1)에서 replica.lag.time.max.ms 이내에 Follower가 fetch하지 않으면, ISR에서 제거한다.
2. Broker 1이 ZooKeeper에게 ISR 변경을 알린다.
3. ZooKeeper는 Partition Metadata에 대한 변경 사항을 Controller에 전달한다.

• Leader가 실패하는 경우, Controller는 Follower 중에서 새로운 Leader를 선출하고 새 Leader와 ISR 정보를 먼저 ZooKeeper에 Push한 다음, 로컬 캐싱을 위해 Broker에 Push한다.

1. ZooKeeper가 session timeout을 통해서 Leader Broker의 장애를 감지한다.
2. Controller가 새로운 Leader를 선출하고 새로운 ISR 리스트를 ZooKeeper에 기록한다.
3. Controller가 모든 Broker에게 새로운 ISR 정보를 Write한다.
4. Client는 메타 데이터를 요청하여 새로운 Leader 정보를 받는다.

---

Broker Failure

* 상황: Broker 4대, Partition 4, Replication Factor 3

Partition 생성 시, Broker들 사이에서 Partition들이 분산하여 배치된다.

Partition 3의 경우 Leader가 죽었기 때문에, Broker 1, Broker 2에 존재하는 Follower Partition 중 하나가 새로운 Leader로 선정된다.

* Partition에 Leader가 없으면, Leader가 선출될 때까지 해당 Partition을 사용할 수 없게 된다.

* Producer의 send()는 retries 파라미터가 설정되어 있으면 계속 재시도를 하지만, 데이터는 전송될 수 없다.

* retries = 0인 경우, NetworkException이 바로 발생한다.

---

Replica Recovery

* 상황 1: acks = all

Leader Partition 0에 장애가 발생하면, 여기서는 Partition 1이 새로운 Leader로 선정되었다고 가정한다. (Partition 1의 M3 데이터는 Commit 전 상태이다.)

1. M3, M4를 Commit 하기 전에 Partition 0에 장애가 발생
2. Partition 1이 새로운 Leader로 선출 -> Leader Epoch가 0에서 1로 증가한다.
3. Partition 2는 M3를 fetch한다.
4. Partition 1은 High Water Mark를 진행한다.
5. Partition 2는 fetch를 다시 수행하고 High Water Mark를 수신하고 진행한다.
6. Producer는 M3, M4에 대한 ACK를 미수신한 상태이므로 M3, M4 송신을 재시도한다.
7. idempotence=false이므로, Partition 1에서 M3는 중복된다.
8. Partition 2는 M3, M4를 fetch한다.
9. Partition 2는 fetch를 다시 수행하고 High Water Mark를 수신하고 진행한다.

* Partition 0이 장애에서 복구된다.

1. Partition 0이 복구되면 ZooKeeper에 연결
2. Partition 0은 Controller로부터 metadata를 받는다.
3. Partition 0은 Partition 1로부터 Leader Epoch를 fetch한다.
4. Partition 0은 Leader가 변경된 시점부터 Truncate(이전 Commit 지점 이후의 데이터를 전부 삭제한다. M3, M4 데이터 삭제)
5. Partition 0은 Partition 1로부터 데이터를 받아 M1, M2, M3, M3, M4 상태로 만들어진다.
6. Partition 0이 ISR 조건에 만족하면 ISRs에 포함된다.

최종 상태

 

* 상황 2: acks = 1

Leader Partition 0에 장애가 발생하면, 여기서는 Partition 1이 새로운 Leader로 선정되었다고 가정한다. (Partition 1의 M3 데이터는 Commit 전 상태이다.)

1. Partition 0에서 장애가 발생하기 전에 Producer는 M4에 대한 ACK를 수신
2. Partition 1이 Leader로 선출된다. -. Leader Epoch가 0에서 1로 증가한다.
3. Partition 2가 M3를 fetch한다.
4. Partition 1은 High Water Mark를 진행한다.
5. Partition 2는 fetch를 다시 수행하고 High Water Mark를 수신하고 진행한다.
6. Producer는 M4에 대한 ACK를 받았으므로, M4 송신을 재시도하지 않아 M4 데이터는 유실된다.

* Availability와 Durability

가용성과 내구성 중 어느 것을 선택할 것인가?

 

** Topic 파라미터 - unclean.leader.election.enable: ISR 리스트에 없는 Replica를 Leader로 선출할 것인지에 대한 옵션(default: false)

해당 옵션이 false일 때, ISR 리스트에 아무 것도 존재하지 않는다면 Leader 선출을 하지 않는다. -> 서비스 중단으로 이어질 수 있다.

해당 옵션을 true로 설정하면, ISR 리스트에 없는 Replica도 Leader로 선출한다. -> 데이터 유실

 

** Topic 파라미터 - min.insync.replicas: 최소 요구되는 ISR의 개수에 대한 옵션(default: 1)

해당 옵션이 1이면, Leader는 반드시 ISR에 존재하므로 오류가 발생하지 않아 보통 2로 사용한다.

ISR이 min.insync.replicas보다 작은 경우, Producer는 NotEnoughReplicas 예외를 수신한다.

Producer에서 acks=all + min.insync.replicas=2로 사용하면 더 강력하게 보장한다.

즉, n개의 Replica가 존재하고, min.insync.replicas=2인 경우 n-2개의 장애까지 허용할 수 있다.

 

데이터 유실을 없게 Kafka를 사용하고자 한다면?

Topic: replication.factor는 2보다 커야 한다.(최소 3이상), min.insync.replicas는 1보다 커야 한다.

Producer: acks는 -1, all 이여야 한다.

 

데이터 유실이 다소 존재하더라도 가용성을 높게 하려면?

Topic: unclean.leader.election.enable를 true로 사용한다.