kafka 리플리케이션, 컨트롤러, 로그

리플리케이션 동작개요

카프카는 브로커의 장애에도 불구하고 연속적으로 안정적인 서비스 제공함으로써 데이터 유실을 방지하며 유연성을 제공합니다. 카프카의 리플리케이션 동작을 위해 토픽 생성 시 유실을 방지하며 유연성을 제공합니다. 카프카의 리플리케이션 동작을 위해 토픽 생성 시 필수값으로 replication factor 라는 옵션을 설정해야 합니다. 

 

# ./kafka-topics.sh  --bootstrap-server localhost:9092 --create --topic peter-overview01 --partitions 1 --replication-factor 3
# ./kafka-topics.sh  --bootstrap-server localhost:9092 --topic peter-overview01 --describe

 

N개의 리플리케이션이 있는 경우 N-1까지의 브로커 장애가 발생해도 메시지 손실없이 안정적으로 메시지를 주고받을 수 있습니다. 이 절에서 생성한 토픽을 기준으로 설명하자면 총 3개의 리플리케이션이 있으므로 그중 2대의 브로커 장애가 발생하더라도 남은 1대의 브로커가 클라이언트들의 요청을 안전하게 처리할 수 있게 됩니다.

 

리더와 팔로워

topic describe 명령어를 실행해보면 출력 내용 중 파티션의 리더라는 부분이 있습니다. 동일한 리플리케이션 이지만 리더만의 역할이 따로 있기 때문입니다. 리더는 리플리케이션 중 하나가 선정되며, 모든 읽기와 쓰기는 그 리더를 통해서만 가능합니다. 다시말해 프로듀서는 모든 리플리케이션에 메시지를 보내는 것이 아니라 리더에게만 메시지를 전송합니다. 팔로워들은 그저 대기만 하는것이 아니라 리더에 문제가 발생하거나 이슈가 있을경우를 대비해 언제든지 새로운 리더가 될 준비를 해야합니다. 따라서 컨슈머가 토픽의 메시지를 꺼내가는 것과 비슷한 동작으로 지속적으로 파티션의 리더가 새로운 메시지를 받았는지 확인하고, 새로운 메시지가 있다면 해당 메시지를 리더로부터 복제합니다. 

 

그럼 리더와 팔로워 중 리플리케이션 동작을 잘하고 있는지 여부 등은 누가 판단하고 어떤기준으로 판단할까요? 리더는 읽고 쓰는 동작은 물론 팔로워가 리플리케이션 동작을 잘 수행하고 있는지도 판단합니다. 만약 팔로워가 특정 주기의 시간만큼 복제 요청을 하지 않는다면 리더는 해당 팔로워가 리플리케이션 동작에 문제가 발생했다고 판단해 ISR그룹에서 추방합니다. ISR내에서 모든 팔로워의 복제가 완료되면 리더는 내부적으로 커밋되었다는 표시를 하게 됩니다. 마지막 커밋 오프셋 위치는 하이워터마크라고 부릅니다. 그리고 이렇게 커밋된 메시지만 컨슈머가 읽어갈 수 있습니다. 

 

복제 유지와 커밋

리더와 팔로워는 ISR이라는 논리적 그룹으로 묶여 있습니다. 이렇게 리더와 팔로워를 별도의 그룹으로 나누는 이유는 기본적으로 해당 그룹안에 속한 팔로워들만이 새로운 리더의 자격을 가질 수 있기 때문입니다. 다시 말해 ISR그룹에 속하지 못한 팔로워는 새로운 리더의 자격을 가질 수 없습니다. 

 

ISR내의 팔로워들은 리더와의 데이터 일치를 유지하기 위해 지속적으로 리더의 데이터를 따라가게 되고, 리더는 ISR내 모든 팔로워가 메시지를 받을 때까지 기다립니다. 하지만 팔로워가 네트워크 오류, 브로커 장애 등 여러 이유로 리더로부터 리플리케이션하지 못하는 경우도 발생할 수 있습니다. 이렇게 뒤쳐진 팔로워는 이미 리더와의 데이터가 불일치한 상태에 놓여 있으므로, 만약 이 팔로워에게 새로운 리더를 넘겨준다면 데이터의 정합성이나 메시지 손실 등의 문제가 발생할 수 있습니다. 따라서 파티션의 리더는 팔로워들이 뒤쳐지지않고 리플리케이션 동작을 잘하고 있는지를 감시합니다. 즉 리더에 뒤처지지 않고 잘 따라잡고 있는 팔로워들만이 ISR그룹에 속하게 되며, 리더에 장애가 발생할 경우 새로운 리더의 자격을 얻을 수 있는 것입니다. 

 

kafka에서는 메시지 불일치 현상을 방지하고자 커민된 메시지만 컨슈머가 읽어 나갈 수 있도록합니다. 이를 보았을때 커밋된 위치가 매우 중요하다는 사실을 알 수 있습니다. 모든 브로커는 재시작될때 커밋된 메시지를 유지하기 위해 로컬 디스크의 replication-offset-checkpoint 라는 파일에 마지막 커밋 오프셋 위치를 저장합니다. replication-offset-checkpoint 파일은 브로커 설정 파일에서 설정한 로그 디렉토리의 경로에 있으며 브로커 설정 파일의 로그 디렉토리는 /tmp/kafka-logs 로 설정되어 있으므로, 해당 디렉토리 하위에 위치합니다. 

 

여타 메시징 시스템들은 리플리케이션 동작에서 리더와 팔로워가 메시지를 잘 받았는지 확인하는 ACK 통신을 하지만, kafka는 ACK 통신 단계를 제거했습니다. 덕분에 kafka의 리더는 메시지를 주고받는 기능에 더욱 집중할 수 있습니다. ACK없이도 리플리케이션이 신뢰할 수 있습니다. 리플리케이션 동작방식은 리더가 push하는 방식이 아니라 팔로워가 pull하는 방식으로 동작합니다.

 

리더에포크와 복구

Leader Epoch는 카프카의 파티션들이 복구 동작을 할때 메시지의 일관성을 유지하기 위한 용도로 이용됩니다. 리더에포크는 컨트롤러에 의해 관리되는 32비트의 숫자로 표현됩니다. 해당 리더에포크 정보는 리플리케이션 프로토콜에 의해 전파되고 새로운 리더가 변경된 후 변경된 리더에 대한 정보는 팔로워에게 전달됩니다. 리더에포크는 복구동작 시 하이워터마크를 대체하는 수단으로 활용됩니다. 

 

리더에포크를 사용하지 않았을때 장애시 메시지를 유실할 수 있는 상황은 다음과 같습니다.

리더의 하이워터마크가 1올라가고 이를 아직 팔로워에 전달하지 못해  팔로워의 하이워커마크가 그대로 인 상태에서 팔로워에 장애가 발생하였다고 가정합시다. 이때 복구된 팔로워는 내부적으로 복구 동작을 하게 됩니다. 

1. 팔로워는 자신이 갖고 있는 메시지들 중에서 자신의 워터마크보다 높은 메시지들은 신뢰할 수없는 메시지로 판단하고 삭제합니다. 

2. 팔로워는 리더에게 최근 오프셋의 새로운 메시지에 대한 가져오기 요청을 보냅니다. 

3. 이 순간 리더였던 브로커가 장애로 다운되면 하이워터마크가 낮은 팔로워가 리더로 승격되면서 새로운 메시지는 유실된다. 

 

이러한 상황때문에 복구동작시 리더에포크를 활용하게 되면 팔로워가 다시 복구된 상황에서 자신의 하이워터마크보다 높은 메시지를 즉시 삭제하지 않고 리더에게 리더에포크 요청을 보냅니다. 

1. 팔로워는 복구동작을 하면서 리더에게 리더에포크 요청을 보냅니다. 

2. 요청을 받은 리더는 리더에포크의 응답으로 x번 오프셋의 메시지까지 라고 팔로워에게 보냅니다. 

3. 팔로워는 자신의 하이워터마크보다 높은 1번 오프셋의 메시지를 삭제하지 않고 리더의 응답을 확인한 후 자신의 하이워터마크를 상향조정합니다. 

 

만약 리더와 팔로우 모두가 동시에 종료되었고 리더가 팔로워보다 늦게 복구된다면 어떻게 될까요?

1. 팔로워였던 브로커가 장애에서 먼저 복구됩니다. 

2. 해당 토픽의 0번 파티션에 리더가 없으므로 팔로워는 새로운 리더로 승격됩니다. 

3. 새로운 리더는 프로듀서로 부터 다음메시지(메시지 1개 유실)를 전달받고 자신의 하이워터마크를 상향조정합니다. 

 

컨트롤러

리더 선출을 맡고 있는 컨트롤러에 대해 알아보자. 카프카 클러스터 중 하나의 브로커가 컨트롤러 역할을 하게 되며 파티션의 ISR 리스트 중에서 리더를 선출합니다. 리더를 선출하기 위한 ISR 리스트 정보는 안전한 저장소에 보관되어 있어야 하는데, 가용성 보장을 위해 주키퍼에 저장되어 있습니다. 

 

1) 파티션 0 의 리더인 1번 브로커가 예기치 않게 다운이 됩니다.
2) 주키퍼는 1번 브로커와의 연결이 끊어진 후, 0번 파티션의 ISR에서 변화가 생겼음을 감지합니다.
3) 컨트롤러는 주키퍼 워치를 통해 0번 파티션에 변화가 생긴것을 감지하고 해당 파티션의 ISR 중에서 새로운 리더를 선출 합니다.
4) 컨트롤러는 0번 파티션의 새로운 리더가 어떤 번호의 브로커가 되었다는 정보를 주키퍼에 기록 합니다.
5) 이렇게 갱신된 정보는 현재 활성화 상태인 모든 브로커에게 전파 됩니다.

 

리더선출과정도 시간이 소요되기 때문에, 파티션 갯수 X 리더선출 소요시간이 총소요 됩니다. 이런 상황을 개선코자 오픈소스 진영은 카프카의 성능개선을 위한 연구를 거듭한 끝에 18년 11월 릴리즈된 kafka 버전 1.1.0부터는 리더 선출 작업속도가 빨라졌습니다 .컨틀루언트 블로그 내용에 따르면 kafka 1.0.0에서 6분 30초 소요되던 작업이 1.1.0에서는 약 3초 만에 완료되었습니다. 

 

제어된 종료의 경우는 어떨까요? 

제어된 종료를 사용하면 카프카 내부적으로 파티션들의 다운타임을 최소화할 수 있습니다. 그 이유는 브로커가 종료되기 전, 컨트롤러는 해당 브로커가 리더로 할당된 전체 파티션에 대해 리더 선출작업을 진행하기 때문입니다. 물론 제어된 종료라 할지라도 리더 선출 작업 시간 동안 일시적으로 다운타임이 발생합니다. 하지만 리더 선출 작업대상 파티션들의 리더들이 활성화된 상태에서 컨트롤러는 순차적으로 하나의 파티션마다 리더를 선출하게 되므로 결과적으로 각 파티션들은 다운타임을 최소화 할수 있습니다. 

 

로그

로그 세그먼트에는 메시지의 내용만 저장되는 것이 아니라 메시지의 키, 벨류, 오프셋, 메시지 크기 같은 정보가 함께 저장되며 로그 세그먼트 파일들은 브로커의 로컬 디스크에 보관됩니다. 하나의 로그 세그먼트 크기가 너무 커져버리면 파일을 관리하기 어렵기 때문에 로그 세그먼트의 최대 크기는 1GB가 기본값으로 설정되어 있습니다. 로그 세그먼트가 1GB보다 커지는 경우에는 기본적으로 롤링 전략을 적용합니다. 다시 말해 하나의 로그 세그먼트에 카프카로 인입되는 메시지들을 계속해서 덧붙이다가 로그 세그먼트의 크기가 1GB에 도달하면 해당 세그먼트 파일을 클로즈 하고 새로운 로그 세그먼트를 생성하는 방식으로 진행됩니다. 

 

kafka에 기본적으로 로그 세그먼트 파일에 대한 롤링 전략이 준비되어 있긴 하지만 kafka 관리자는 1GB크기의 로그 세그먼트 파일이 무한히 늘어날 경우를 대비해 로그 세그먼트에 대한 관리 계획을 수립해둬야 합니다. 로그 세그먼트를 관리하는 방법은 크게 로그 세그먼트 삭제와 컴팩션으로 구분할 수 있습니다.

 

가) 로그세그먼트 삭제

로그 세그먼트 삭제 옵션은 브로커의 설정파일인 server.properties 에서 log.cleanup.policy가 delete로 명시되어야 합니다. 이 값은 Default이므로 관리자가 server.properties에 해당 옵션을 따로 명시하지 않았다면 로그 세그먼트는 삭제 정책이 적용됩니다. 

 

# ./kafka-topics.sh  --bootstrap-server localhost:9092 --create --topic peter-test03 --partitions 1

# ./kafka-console-producer.sh --bootstrap-server localhost:9092 --topic peter-test03

# ./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic peter-test03 --from-beginning

# ./kafka-configs.sh --bootstrap-server localhost:9092 --topic peter-test03 --add-config retention.ms=0 --alter

# ./kafka-topics.sh  --bootstrap-server localhost:9092 --topic peter-test03 --describe

# ls /tmp/kafka-logs/peter-test03-0/
//5분정도 뒤에 조회를 해보면 00000000000000001로 시작하는 로그파일이 보임

# ./kafka-configs.sh --bootstrap-server localhost:9092 --topic peter-test03 --delete-config retention.ms=0 --alter

 

 

나) 로그세그먼트 컴팩션

로그를 삭제하지 않고 컴펙션하여 보관할 수 있습니다. 현재 활성화된 세그먼트는 제외하고 나머지 세그먼트들을 대상으로 컴펙션이 실행됩니다. 컴펙션 할지라도 카프카의 로컬 디스크에 로그를 무기한 보관한다면 로그의 용량은 감당할 수없이 커져서 디스크가 저장할 수 있는 용량의 한계에 도달할 것입니다. 따라서 kafka에서는 단순하게 메시지를 컴펙션하여 보관하기 보다는 좀더 효율적인 방법으로 컴펙션 합니다. 카프카에서 로그 세그먼트를 컴펙션하면 메시지의 키값을 기준으로 마지막의 데이터만 보관하게 됩니다. 

 

메시지의 키값을 기준으로 컴팩션하는 방법이 다소 생소할 수 있는데 로그 컴팩션 기능을 이용하는 대표적인 예제는 바로 카프카의 __consummer_offset 토픽입니다. __consummer_offset 토픽은 카프카의 내부 토픽으로 컨슈머 그룹의 정보를 저장하는 토픽입니다. 컨슈머 그룹의 중요한 정보는 해당 컨슈머 그룹이 어디까지 읽었는지를 나타내는 오프셋 커밋 정보인데 __consumer_offset에 키와 벨류 형태로 메시지가 저장됩니다. 

 

컴팩션 전 로그에서 키 값이 K1인 벨류들을 확인해보면 오프셋 0일때 V1, 오프셋 2일때 V3, 오프셋 3일때 V4로 확인됩니다. 카프카의 메시지들은 오프셋순으로 저장되므로 K1 키값의 마지막 벨류는 V4임을 알 수 있습니다. 

따라서 컴팩션 후 로그를 보면 K1키의 벨류는 V4입니다. 다시 정리해보면 K1키로 들어온 메시지들이 모두 저장되어 있다가 로그 컴팩션을 통해 V1, V3는 삭제되고 가장 마지막의 메시지인 V4만 남아 로컬 디스크에 저장됩니다. 

 

로그컴팩션의 장점은 바로 빠른 장애 복구입니다. 장애 복구시 전체 로그를 복구하지 않고 메시지의 키를 기준으로 최신의 상태만 복구합니다. 따라서 전체 로그를 복구할 때보다 복구 시간을 줄일 수 있다는 장점이 있습니다. 

 

하지만 빠른 재처리라는 장점이 있다고 해서 모든 토픽에 로그 컴팩션을 적용하는 것은 좋지 않습니다 .키 값을 기준으로 최종값만 필요한 워크로드에 적용하는 것이 바람직 합니다. 그리고 kafka에서 로그 컴팩션 작업이 실행되는 동안 브로커의 과도한 입출력 부하가 발생할 수 있으니 유의해야 합니다. 따라서 반드시 브로커의 리소스 모니터링도 병행하여 로그 컴팩션을 사용하기를 권장합니다.