Kafka 클러스터 구성 및 장애 해결

1) 서론

주키퍼(zookeeper) 1개 + 카프카(kafka) 3개 구성의 클러스터를 구성합니다. 

 

학습에 목적을 두고 있기 때문에 최대한 간단한 클러스터를 구성했습니다. 이를 통해 카프카, 주키퍼의 설정 방식을 공부합니다.

 

+) 학습과정의 블로그이므로, 오류는 댓글로 남겨주시면 감사히 수정하겠습니다!

 

---

2) Zookeeper

1대의 주키퍼 서버를 사용한다면, 더 설정할 것은 없습니다. 

 

아래의 명령어를 통해 주키퍼를 먼저 실행합니다. 

$ bin/zookeeper-server-start config/zookeeper.properties

or 

$ brew services start zookeeper

---

3) Kafka

카프카 브로커의 클러스터를 구성합니다.

 

우선, config/server.properties를 복사하여 server[숫자].properties로 만듭니다. 이때 변경해야 할 것은 아래와 같습니다

// ex) 포트: 8081 ~ 8083
listeners=PLAINTEXT://0.0.0.0:[포트]
advertised.listeners=PLAINTEXT://localhost:[포트]

// ex) brokder.id=1
broker.id=[번호]

// 각 카프카 서버가 데이터를 저장할 공간
log.dirs=/usr/local/var/lib/kafka-logs/broker[번호]

// 각 카프카 서버가 메세지 데이터를 보관하는 기간 (옵션)
// log.retention.hours가 존재할 경우 ms가 우선순위
log.retention.ms=50000

// 통신할 주키퍼 서버
zookeeper.connect=localhost:2181

• 각 브로커마다 별도의 log.dirs가 필요

 

---

4) Topic

4.1) Topic 생성

$ bin/kafka-topics --create --zookeeper localhost:2181 
(연결)
--replication-factor 3 --partitions 10 --topic test

• Topic: test
• replication-factor
  • 3 
  • 메시지 데이터 복사할 브로커 개수 

• partitions
  • 10
  • 메세지를 병렬적으로 나누어서 전송

 

4.2) 토픽 정보

만들어진 토픽을 아래의 명령어로 확인합니다.

// 해당 주키퍼 포트의 토픽을 확인합니다.
$ kafka-topics --describe --zookeeper localhost:2181

Topic: test	TopicId: a32WENMJRXe3Q3en1Ok1hg	PartitionCount: 10	ReplicationFactor: 3	Configs: 
	Topic: test	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2
	Topic: test	Partition: 1	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
	Topic: test	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1

    ... 생략

이때 중요한것은 각 파티션마다 리더(leader)가 있고, 복제본(replicas)이 있는데요.

 

각 파티션들은 어떤 브로커가 리더가 되는지 알 수 없습니다.

 

리더의 선출은 컨트롤러(브로커)에 의해서 선출됩니다. 선출되지 않은 나머지 카프카 브로커들은 팔로워(follower)가 됩니다. 그리고 해당 파티션의 리더에 장애가 발생한 경우 새로운 리더를 선출하고, 나머지는 팔로워가 됩니다.

 

이때 오직 리더에게만 read, write를 할 수 있는데요. 리더에 작성된 데이터를 팔로워들이 복사하는 과정을 거칩니다.

 

이처럼 파티션은 리더, 팔로워를 선출하고 복제본을 만드는 과정을 거칩니다. 안전하고, 모든 것이 완벽해 보이는데요.

 

하지만 파티션을 무조건 많이 만드는것이, 꼭 좋은 것은 아닙니다. 

 

위와 같이 선출, 복제본 생성의 과정을 거치기 때문에 지나칠 경우 성능적으로 문제가 발생한다고 합니다. 그리고 브로커가 토픽의 안정성을 꾸준히 체크하게 되는데, 이러한 과정에서 자원을 소모할 수 있습니다. 

 

---

5) 브로커 장애 발생 시

카프카의 브로커는 주키퍼와 다르게, 새로운 리더 선출에 "2f + 1"개의 복제본이 필요하지 않습니다. 

 

ISR이라는 방식을 사용하고 있습니다. 이는 In -Sync Replica(동기화되고 있는 상태인 복제본) 그룹니다. 

 

최초에는 Replica 수 = ISR 수입니다. 그리고 파티션 리더 또한 복제본에 포함됩니다.

 

리더는 팔로워들이 데이터를 잘 복제하고 있는지 주기적으로 확인합니다. 만약 하나의 팔로워에 장애가 생기거나, 리더의 데이터를 제대로 업데이트하지 못한다면 리더에 의해 ISR 그룹에서 퇴출됩니다. ISR 그룹에서 퇴출된 브로커가 있다면 Replica 수 /= ISR 수가 됩니다. 즉 반드시 ISR 그룹과 복제본 그룹의 수가 같은 필요 없습니다. 

 

하지만 리더에 장애가 발생하면, 어떻게 될까요?

 

이때는 리더 재선출 과정을 거쳐야 하는데요. 리더 재선출은 컨트롤러에 의해서 이루어집니다. 컨트롤러가 ISR의 가장 앞에 있는 브로커를 리더로 선정합니다. 이때 중요한 것은 새 리더는 오직 ISR의 그룹에 포함된 브로커만 될 수 있습니다. 왜냐하면 현재 데이터를 sync 하고 있지 않는 브로커를 리더로 하면, 데이터의 손실이 발생할 수 있기 때문입니다. 

 

그렇다면 컨트롤러 브로커는 어떻게 장애가 발생한지 알 수 있을까요? 

 

모든 브로커는 실행 시 주키퍼 서버에 컨트롤러 노드(/controller)로 등록되기 위한 시도를 합니다. 이때 가장 먼저 성공적으로 컨트롤러 노드로 등록된 브로커가 컨트롤러 노드가 됩니다. 

 

브로커에 장애 발생 시 주키퍼는 등록된 컨트롤러 노드에게 알려줍니다. 그리고 컨트롤러는 장애를 확인하고 새로운 리더를 뽑고, ISR 그룹을 관리합니다. 

 

하지만 만약에 ISR 그룹이 비어있다면 어떻게 새 리더를 선정할까요? 즉 현재 sync 중인 브로커가 없을 때를 가정합니다.

 

카프카의 기본 설정은 가장 최근의 리더가 복구되는 것을 기다리는 것입니다. 이는 가장 최근까지 리더였던 브로커의 데이터를 기반으로 하기 때문에 데이터의 유실을 방지할 수 있습니다. 다만 복구에 많은 시간이 필요할 수도 있습니다. 언제 이 리더가 복구될지 장담할 수 없으니까요.

 

다른 방법으로는 "unclean.leader.election.enable = true" 설정도 있습니다. 이는 가장 빨리 회복되는 브로커를 리더로 선정합니다. 장애 복구 시간을 줄일 수 있다는 장점이 있지만, 데이터의 유실이 발생하는 단점도 있습니다. 왜냐하면 가장 빨리 회복되는 브로커가 어디까지의 데이터를 복제했는지 장담할 수 없기 때문입니다. 

 

// 브로커 1번 종료
Topic: test	TopicId: a32WENMJRXe3Q3en1Ok1hg	PartitionCount: 10	ReplicationFactor: 3	Configs: 
	Topic: test	Partition: 0	Leader: 3	Replicas: 3,1,2	Isr: 3,2
	Topic: test	Partition: 1	Leader: 2	Replicas: 1,2,3	Isr: 2,3
	Topic: test	Partition: 2	Leader: 2	Replicas: 2,3,1	Isr: 2,3

    ... 생략

• 브로커 1번 장애 발생
• 새로운 리더 선출
• 브로커 1번 ISR 퇴출

---

6) 카프카 리밸런싱 설정 변경

카프카 리밸런싱에는 "auto.leader.rebalance.enable" 설정을 활용합니다. 

 

위에서 언급했듯이 주키퍼 컨트롤러(브로커)가 주기적으로 스레드를 활용해 리밸런싱 여부를 확인합니다. 그리고 이때 필요하다면, 자동으로 리밸런싱 하여 리더를 선출합니다. 기본 설정은 true입니다. 

 

컨트롤러는 브로커인데요. 클러스터의 여러 브로커 중 하나의 브로커가 선출됩니다. 그리고 이 컨트롤러가 파티션의 리더를 재선출하기도 하며, 복제본의 상태를 확인하기도 합니다. 브로커들의 클러스터 전체를 관리하는 역할을 합니다. 컨트롤러는 주키퍼의 znode에 의해 선출됩니다. 

 

하지만 주기적으로 리밸런싱이 필요한지 확인하는 것은 자원 소모가 필요합니다. 그래서 "leader.imbalance.check.interval.seconds" 통해서 확인 주기를 설정해줄 수 있습니다. 기본 설정 시간은 5초입니다. 

 

위와 비슷하게 "leader.imbalance.per.broker.percentage"도 있습니다. 이는 불안정함이 설정된 퍼센트가 넘으면, 리밸런싱을 하는것입니다. 기본은 10%입니다.

6. 1) 설정

server[번호].properties

group.initial.rebalance.delay.ms=3000
auto.leader.rebalance.enable=true

leader.imbalance.check.interval.seconds=30
// or 
leader.imbalance.per.broker.percentage=5

• group.initial.rebalance.delay.ms: 리밸런싱 시 새로운 컨슈머 추가를 대기하는 시간
• auto.leader.rebalance.enable: 파티션 리더 자동 리밸런싱 여부 
• leader.imbalance.check.interval.seconds: 컨트롤러에 의해 리밸런싱 여부 확인 시간
• leader.imbalance.per.broker.percentage: 컨트롤러에 의해 리밸런싱 여부 확인 퍼센트

 

6. 2) Test

// 1번 브로커 종료
Topic: start	TopicId: 3J7QAB2MQjGmsDiiCVd-Lw	PartitionCount: 10	ReplicationFactor: 3	Configs: 
	Topic: start	Partition: 0	Leader: 2	Replicas: 1,2,3	Isr: 2,3
	Topic: start	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3
	Topic: start	Partition: 2	Leader: 3	Replicas: 3,1,2	Isr: 3,2
	Topic: start	Partition: 3	Leader: 3	Replicas: 1,3,2	Isr: 3,2

당연하게 파티션의 리더가 리밸런싱 됩니다.

 

 

7) 참고 문헌

https://kafka.apache.org/documentation/#replication

https://stackoverflow.com/questions/41334691/leader-election-in-zookeeper-and-kafka

https://docs.confluent.io/platform/current/installation/configuration/broker-configs.html