카프카의 중요 내부 로직을 분석하고 정리해 보고자 한다. 이번 글에서는 컨트롤러에 대해서 살펴보자.
그럼 컨트롤러란 무엇인가? 컨트롤러의 역할을 먼저 살펴보고 동작방식을 분석하자.
1. 컨트롤러란 무엇인가?
- 클러스터에서 하나의 브로커가 컨트롤러 역할을 한다.
- 브로커의 상태 체크.
- 죽은 브로커가 담당한 파티션의 새 리더 선출.
- 새롭게 선출된 리더 정보를 모든 브로커에 전달.
이름처럼 컨트롤러는 브로커들을 관리한다. 브로커가 정상적인지 상태를 체크하며 죽은 브로커가 있을 경우, 해당 브로커가 가지고 있던 파티션 리더들을 재분배 한다.
카프카는 데이터의 등록/ 소비를 파티션 리더가 모두 담당하므로 브로커의 상태 체크가 원활하지 않을 경우 카프카에 심각한 위험요소가 된다.
컨트롤러는 여러 브로커 중에 하나의 브로커가 담당한다.
2. 컨트롤러 선정
어떤 어떤 브로커에게 컨트롤러 역할을 맡기게 될까?
컨트롤러 선정은 주키퍼의 임시노드(ephemeral node)를 활용한다. 브로커가 주키퍼 임시노드 /controller에 데이터가 없으면 브로커 번호를 저장하고 컨트롤러 역할을 담당한다.
다른 브로커들이 주키퍼 임시노드 /controller를 확인했을 때는 데이터가 있으므로 저장된 브로커 번호를 컨트롤러로 인식한다.
이런 방식으로 전체 브로커에서 하나의 브로커만 컨트롤러 역할을 담당하게 선정한다.
그럼 컨트롤러 역할을 하는 브로커 죽으면 어떻게 다시 컨트롤러를 선정할까?
컨트롤러가 죽으면 주키퍼의 임시노드 /controller가 삭제되고 다른 브로커들은 주키퍼로부터 삭제 이벤트를 받게 된다.
[Code-1]은 주키퍼 임시노드 /controller의 데이터 삭제/ 변화가 생겼을 때 이벤트를 받도록 설정하는 코드이다.
주키퍼 임시노드가 삭제됐을 때는 Reelect 이벤트, 데이터 변화가 생겼을 때는 ControllerChange 이벤트를 발생시킨다.
class ControllerChangeHandler(eventManager: ControllerEventManager) extends ZNodeChangeHandler {
override val path: String = ControllerZNode.path
override def handleCreation(): Unit = eventManager.put(ControllerChange)
override def handleDeletion(): Unit = eventManager.put(Reelect)
override def handleDataChange(): Unit = eventManager.put(ControllerChange)
}
[Code-1. 주키퍼 컨트롤러 이벤트 핸들러]
주키퍼 임시노드가 삭제되었을 때 새로운 컨트롤러의 선출하는 로직은 [Code-2]와 같이 elect() 메서드가 수행한다.
private def elect(): Unit = {
activeControllerId = zkClient.getControllerId.getOrElse(-1)
if (activeControllerId != -1) {
return
}
try {
val (epoch, epochZkVersion) = zkClient.registerControllerAndIncrementControllerEpoch(config.brokerId)
controllerContext.epoch = epoch
controllerContext.epochZkVersion = epochZkVersion
activeControllerId = config.brokerId
onControllerFailover()
} catch {
case e: ControllerMovedException =>
maybeResign()
case t: Throwable =>
triggerControllerMove()
}
}
[Code-2. 컨트롤러 선출]
[Code-2]의 내용을 보면은 간단해 보인다. 9번 라인에서 주키퍼에 현재 브로커의 번호를 등록하는 방법으로 선출을 완료한다.
동시에 여러 브러커가 컨트롤러로 할당받기 위해 경쟁하게 되는데 선출되지 않은 브로커들은 Exception을 발생시켜 컨트롤러가 처리해야 하는 핸들러들을 리셋한다.
주의해야 할 부분은 epoch 값들이 보이는데 이것은 주키퍼에 데이터를 넣을 때 유효한 요청인지 확인할 때 쓰는 값이다. 이것은 카프카 버전 업그레이드 과정에서 개선된 내용을 기술할 때 추가 설명 하도록 하겠다.
3. 컨트롤러 개선 포인트
카프카 버전이 올라가면서 컨트롤러에 크게 보면 2가지의 성능 개선이 있었다. 컨트롤러 종료/ 재구동시 시간 지연과 컨트롤러 failover(시스템 대체 작동) 이슈이다.
3.1. 컨트롤러 종료 이슈
컨트롤러의 종료 이후, 새로운 컨트롤러가 선정이 되면 컨트롤러는 새로 선출된 파티션 리더들의 정보를 주키퍼와 다른 브로커들에게 전달하게 된다.
아래 그림에서 3번, 4번을 나타낸다. 이 때, 성능 이슈가 있었다. 3번, 4번 단계에서 다음과 같은 속도 저하 원인이 있었다.
- 3번
- 주키퍼에 파티션 리더 정보를 순차적으로 저장한다(한 파티션 저장 끝나고 다음 파티션 저장).
- 주키퍼에 정보를 저장하는데 시간이 오래 걸렸다.
- 4번
- 각 파티션 단위로 하나의 요청을 만들었다.
- 브로커에 작은 요청이 무수히 쌓이게 되어서 브로커가 처리할 때 CPU 이슈가 발생할 수 있었다.
결국 속도 저하의 원인은 작은 단위의 변경 요청을 한번에 보내지 않고 여러번 보내서 발생한 문제였다. 그래서 개선 방법도 아이디어는 간단했다.
주키퍼와 다른 브로커와 통신할 때 요청을 한번에 묶어서 보냈다.
3.2. 컨트롤러 failover(시스템 대체 작동)
다음 그림은 컨트롤러 failover시에 발생할 수 있는 좀비 컨트롤러 이슈를 보여준다.
좀비 컨트롤러 이슈가 발생하는 이유는 새로운 컨트롤러가 선출된 이후에 주키퍼의 메타데이터를 가져오는 시간이 필요하기 때문에 발생한다.
새로운 컨트롤러 로딩되는 시간에 과거 컨트롤러가 죽어있으면 상관이 없는데 죽지 않고 살아있는 상태였더라면 문제가 발생한다.
다른 브로커들이 과거 컨트롤러와 통신을 진행할 것이기 때문이다. 이와 같은 좀비 컨트롤러가 발생할 수 있는 것은 GC 등으로 오랜 시간 stop the world가 생겼을 때 컨트롤러가 죽었다고 판단하기 때문이다.
그럼 이와 같은 현상은 어떻게 방어할 수 있을까?
컨트롤러의 버전을 만들어서 통신할 때 과거 버전의 컨트롤러를 방어하는 로직을 만들면 된다.
그래서 앞서 설명한 epoch가 이 로직에 활용된다.
주키퍼와 통신을 할 때에도 버전 체크 로직이 들어가게 된다.
주키퍼에 데이터를 변경하기 위한 부분에서 아래과 같이 version 정보를 가진 SetData 클래스를 활용하였다.
2가지의 큰 개선으로 인해서 카프카는 재시작, failover 시에 새로운 컨트롤러로 로딩되는 시간이 현저하게 감소하였다.
참고문서
반응형
'Big Data > Kafka' 카테고리의 다른 글
| [Kafka] Introducing ksqlDB (0) | 2019.11.21 |
|---|---|
| [Kafka] Sink Connector flush 분석 (0) | 2019.10.08 |
| [Kafka] mirrorMaker v1 단점. v2는? (0) | 2019.07.18 |
| [Kafka] Configurable SASL callback handler (0) | 2019.04.18 |
| [Kafka] 파티션 이동 (0) | 2019.04.12 |