실전 Kafka 개발부터 운영까지. 6장 컨슈머의 내부 동작 원리와 구현

실전 카프카 개발부터 운영까지

컨슈머의 주요한 역할: 카프카에 저장된 메시지를 가져오는 것

6.1 컨슈머 오프셋 관리

오프셋이란? 
메시지의 위치, 숫자 형태로 나타냄

컨슈머 동작 중 가장 핵심: 오프셋 관리

컨슈머가 메시지를 꺼내오므로 어디까지 가져왔는지 표시하는 것이 가장 중요하기 때문. 복구할 때도 사용

컨슈머 그룹이 자신의 오프셋 정보를 카프카에서 가장 안전한 저장소인 토픽에 저장

컨슈머 그룹에 컨슈머 1과 컨슈머2라는 2개의 컨슈머가 존재할 때

컨슈머들은 지정된 토픽의 메시지를 읽은 뒤, 읽어온 위치의 오프셋 정보를 __consumer_offsets에 기록

기록된 정보를 바탕으로 컨슈머 그룹은 자신의 그룹이 속해 있는 컨슈머의 변경이 발생하는 경우(컨슈머 장애 또는 이탈) 해당 컨슈머의 다음으로 읽어야할 위치를 추적할 수 있음

offsets.topic.num.partitions: 기본값 50

offsets.topic.replication. factor: 기본값 3

파티션 수와 리플리케이션 팩터 수는 기본값으로 사용해도 충분

가끔 리플리케이션 팩터 수가 1로 설정되어 있는 경우가 있으므로 주의

6.2 그룹 코디네이터

컨슈머들은 하나의 커슈머 그룹의 구성원에 속함.

컨슈머 그룹 내의 각 컨슈머들은 자신의 정보를 공유하며 하나의 공동체로 동작

컨슈머 그룹 내의 컨슈머들은 본인 그룹을 언제든지 떠날 수 있고, 새로운 컨슈머가 합류할 수도 있음

그러므로 컨슈머 그룹은 각 컨슈머들에게 작업을 균등하게 분배해야함

컨슈머 리밸런싱(컨슈머 리밸런싱이 일어났다): 컨슈머 그룹에서 컨슈머들에게 작업을 균등하게 분배하는 동작

5장에서의 정확히 한 번 정송을 관리하는 것을 트랜잭션 코디네이터

컨슈머 그룹에서도 그룹 관리를 위해 존재하는데 이를 그룹 코디네이터라고 함

그룹 코디네이터의 목적은 컨슈머 그룹이 구독한 토픽과 파티션과 그룹의 멤버들을 트래킹하는 것

따라서 그룹의 멤버에 변화가 생기면 작업을 균등하게 재분배하기 위해 컨슈머 리밸런싱이 동작

그룹 코디네이터는 컨슈머 그룹별로 존재, 카프카 클러스터 내의 브로커 중 하나의 위치

컨슈머 그룹이 브로커에 최초 연결 요청시 브로커 중 하나에 그룹 코디네이터가 생성됨

그룹 코디네티어는 컨슈머 그룹의 컨슈머 변경과 그룹의 멤버 변경시 변경된 내용을 컨슈머에게 전달

컨슈머 그룹 등록 과정

1. 컨슈머는 컨슈머의 설정값 중에서 bootstrap.brokers 리스트에 있는 브로커에게 컨슈머 클라이언트와 초기 커넥션을 연결하기 위한 요청을 보냄
2. 해당 요청을 받은 브로커는 그룹 코디네티어를 생성하고 컨슈머에게 응답을 보냄. 컨슈머 그룹의 첫 번째 컨슈머가 등록될 때까지 아무 작업도 일어나지 않음
3. 그룹 코디네이터는 group.initial.rebalance.delay.ms의 시간 동안 컨슈머의 요청을 기다림
4. 컨슈머는 컨슈머 등록 요청을 그룹 코디네이터에게 보냄. 이때 가장 먼저 요청을 보내는 컨슈머가 컨슈머 그룹의 리더가 됨
5. 컨슈머 등록 요청을 받은 코디네이터는 해당 컨슈머 그룹이 구독하는 토픽 파티션 리스트 등 리더 컨슈머의 요청에 응답을 보냄
6. 리더 컨슈머는 정해진 컨슈머 파티션 할당 전략에 따라 그룹 내 컨슈머들에게 파티 션을 할당한 뒤 그룹 코디네이터에게 전달
7. 그룹 코디네이터는 해당 정보를 캐시하고 각 그룹 내 컨슈머들에게 성공을 알림
8. 각 컨슈머들은 각자 지정된 토픽 파티션으로부터 메시지들을 가져옴

단순히 컨슈머에서 bootstrap.brokers와 group.id만 설정하여 컨슈머 그룹을 생성하고 카프카로부터 구독한 메시지를 읽어오는 것이 아닌

실제로는 컨슈머 그루보가 그룹 코디네이터가 서로 긴밀하게 내용을 주고 받음

커슈머 그룹은 그룹 코디네이터와 연결되어 관리를 받음

컨슈머들은 자신들이 속한 컨슈머 그룹에서 합류할 때는 join 요청을, 떠날때는 leave 요청을 보냄

컨슈머들의 변경을 갑지하기 위해 그룹 코디네이터와 컨슈머들은 서로 히트비트를 주고 받음

컨슈머가 문제있을 경우 리밸런싱 동작을하여 전체 균형을 다시 맞춤

히트비트 말고 할당된 파티션에서 컨슈머가 정상적으로 메시지를 가져가고 있는지 poll() 동작 여부를 통해 확인할 수도 있음

컨슈머 리밸런싱은 높은 비용을 사용할 수 있으므로 리밸런싱은 지양하는 것이 좋음

옵션 값을 사용하여 컨슈머 다운을 적절히 조절하여 컨슈머의 타임아웃이나 일시적인 TCP 패킷 손실을 방지해 원하지 않은 리밸런싱을 방지할 수 있음

6.3 스태틱 멤버십

컨슈머의 재시작으로 인해 리밸런싱이 일어나는 상황

일반적인 컨슈머 그룹 동작에서 각 컨슈머를 식별하기 위해 엔티티 ID를 부여하게 됨

재시작시 새로운 컨슈머로 인식, 새로운 ID가 부여되어 리밸런싱 발생

불필요한 재시작을 방지하기 위한 스태틱 멤버십

스태틱 멤버십이 적용된 컨슈머는 그룹에서 떠날 때 그룹 코디네이터에게 알리지 않아 리밸런싱이 일엉나지 않음

재시작시 시작할 때 1번, 떠날 때 1번 피할 수 있음

기본값이 null string인 group.instance.id를 설정시 스태틱 멤버십이 적용됨

중복을 허용하지 않으므로 접두어를 작성하는 것이 좋음

또한 session.timeout.ms를 기본값보다 크게 조정하여 코디네이터가 하트비트를 못 받았을 경우를 대비해 불필요한 리밸런싱 방지

5.리밸런싱동작.png

컨슈머의 리밸런싱은 재시작되거나 그룹에서 떠나는 컨슈머만 대상으로 당작하는 것이 아닌, 커슈머 그룹 내 전체 컨슈머를 대상으로 동작함

컨슈머 리밸런싱 동작 과정 중 일시적으로 모든 컨슈머가 일시 중지하게 되므로 불필요한 리밸런싱은 최대한 줄여야함

6.4 컨슈머 파티션 할당 전략

프로듀서의 파티셔너: 레코드를 토픽의 어느 파티션으로 전송할지 결정하는 역할

컨슈머의 동작에서도 이와 유사하게 대상 토픽의 어느 파티션으로부터 레코드를 읽어올지 결정함

컨슈머의 구릅 리더 컨슈머가 정해진 파티션 할당 전략에 각 커슈머와 대상 토픽의 파티션을 매칭시킴

파티션 할당 전략은 컨슈머 옵션의 partition.assignment.strategy로 표시

1. RangeAssignor(레인지 전략)
2. RoundRobinAssignor(라운드로빈 전략)
3. StickyAssignor(스티키 전략)
4. CooperativeStickyAssignor(협력적 스티키 전략)

6.4.1 레인지 파티션 할당 전략

파티션 할당 전략 중 기본 값. 각 토픽별로 할당 전략을 사용

먼저 구독하는 토픽에 대한 파티션을 순서대로 나열한 후 컨슈머를 순서대로 정렬

그 후 각 컨슈머가 몇 개의 파티션을 할당해야 하는지 전체 파티션 수를 컨슈머 수로 나눔

컨슈머 수와 파티션 수가 일치하면 균등하게 할당되지만 균등하게 나눠지지 않는 경우 앞쪽의 컨슈머들은 추가 파티션을 할당받음

토픽1과 토픽2는 각각 파티션 3개로 이루어져 있음

토픽1과 토픽2는 3개이므로 3/2를 하여 균등하게 분배되지 않으므로 먼저 정렬된 컨슈머 1에 1개를 추가배치하여 결과적으로 컨슈머1에 4, 컨슈머2에 4로 나눔

컨슈머에 균등하게 파티션이 분배되지 않을 수 있으므로 주의

6.4.2 라운드 로빈 파티션 할당 전략

파티션 할당 전략 중 가장 간단한 할당 방식

모든 파티션과 컨슈머 그룹 내 모든 컨슈머를 나열한 후 라운드로빈으로 하나싞 파티션과 컨슈머를 할당하는 전략

8.라운드로빈파티션할당전략.png

매우 균등하게 매핑

6.4.3 스티키 파티션 할당 전략

리밸런싱 동작으로 인해 파티션이 재할당 될 때 기존에 매핑됐던 파티션과 동일한 커슈머가 다시 매핑됨을 보장할 수 없음

재할당 작업이 발생해도 기존에 매핑됐던 파티션과 컨슈머를 최대한 유지하는 전략

첫 번째 목적: 가능한 균형 잡힌 파티션 할당

두 번째 목적: 재할당 발생시 되도록 기존의 할당된 파티션 정보를 보장

두 번째 목적보다 첫 번째 목적의 우선순위가 더 높음

그러므로 무조건 기존의 파티션과 컨슈머를 유지하는 것은 아님. 새로운 컨슈머와 연결될 수도 있음

컨슈머2 제외 전

컨슈머2 제외 된 상황

1. 컨슈머2가 컨슈머 그룹에서 떠남
2. 리밸런싱 동작
3. 모든 파티션을 순서대로 배치(토픽1-파티션0, 토픽1-파티션1, 토픽2-파티션0, 토픽2-파티션1, ...)
4. 모든 컨슈머를 순서대롭 ㅐ치(컨슈머1, 컨슈머3)
5. 라운드 로빈 파티션 할당 전략에 맞춰 하나씩 매핑

리밸런싱 작업 후

컨슈머2에 할당된 파티션들만 컨슈머1과 컨슈머3에 각각 할당됨을 알 수 있음

• 컨슈머들의 최대 할당된 파티션의 수 차이는 1
• 기존에 존재하는 파티션 할당은 최대한 유지함
• 재할당 동작시 유효하지 않은 모든 파티션 할당은 제거
• 할당되지 않은 파티션들은 규현을 맞추는 방법으로 컨슈머들에 할당

균형을 맞추고 기존 컨슈머에 할당된 파티션을 최대한 유지하여 컨슈머에 새로 할당하는 파티션 수를 최소화함

6.4.4 협력적 스티키 파티션 할당 전략

스티키 파티션 할당 전략과 동일하지만 컨슈머 그룹 내부의 리밸런싱 동작이 고도화된 전략

컨슈머 리밸런싱 동작시 내부적으로 EAGER라는 리밸런스 프로토콜을 사용, EAGER 프로콜은 컹슈머 리밸런싱 동작 시 컨슈머에 할당된 모든 파티션을 항상 취소함

그러한 이유

1. 컨슈머들의 파티션 소유권 변경

A 컨슈머가 소유하는 0번 파티션의 소유권을 B 컨슈머에게 할당해야할 때 둘 이상의 컨슈머가 동일한 파티션을 소유할 수 없으므로 0번 파티션의 소유권을 B 컨슈머에게 넘길 수 없음

2. 그룹 내에서 여러 파티션들에 대해 소유권 변경 작업(새로운 파티션 할당 작업)이 동시에 이뤄줘야하므로 로직을 단순하게 구현하기 위해

일반적인 리밸런싱 동작 과정

리밸런싱 도중에도 프로듀서는 해당 토픽을 타깃으로 지속적으로 메시지를 전송함

컨슈머의 다운타임 동안 LAG이 급격하게 증가됨

세번째 단계에서 구독한 파티션이 컨슈머들에게 재할당됨

스태틱 멤버십의 기능을 통해 불필요한 리밸런싱 동작이 일어나지 않게 할 수는 있지만 불가피하기 리밸런싱이 발생시 여전히 컨슈머의 다운타임이 발생함

컨슈머의 다운타임은 대량의 메시지를 컨슘하는 경우에 매우 큰 부담이 됨

2.4 버전부터 협력적 스티키는 내부 리밸런싱 프로토콜인 EAGER(이거)가 아닌 COOPERACTIVE(협력적) 프로토콜 적용

리밸런싱이 동작하기 전의 컨슈머 상태를 유지할 수 있게함

기존의 리밸런싱 동작

한 번에 모든 파티션 할당 작업이 끝나지만 전체 컨슈머가 일시적으로 멈춘 상태에서 리밸런싱이 이루어짐

새로운 스티키 할당 전략

동작 중인 커슈머들에게 영향을 주지 않은 상태에서 몇 차례에 걸쳐 리밸런싱이 이루어짐

1. 컨슈머 그룹에 peter-kaka01이 합류하면서 리밸런싱이 트리거됨 (① 감지 단계)

2. 컨슈머 그룹 내 컨슈머들은 그룹 합류 요청과 자신들이 컨슘하는 토픽의 파티션 정보(소유권)를 그룹 코디네이터로 전송 (① 감지 단계)

3. 그룹 코디네이터는 해당 정보를 조합해 컨슈머 그룹의 리더에게 전송 (① 감지 단계)

4. 컨슈머 그룹의 리더는 현재 컨슈머들이 소유한 파티션 정보를 활용해 제외해야 할 파티션 정보를 담은 새로운 파티션 할당 정보를 컨슈머 그룹 멤버들에게 전달 (② 첫 번째 리밸런싱 단계)

5. 새로운 파티션 할당 정보를 받은 컨슈머 그룹 멤버들은 현재의 파티션 할당 전략과 차 이를 비교해보고 필요 없는 파티션을 골라 제외시킴. 이전의 파티션 할당 정보와 새로운 파티션 할당 정보가 동일한 파티션들에 대해서는 어떤 작업도 수행하지 않음 (② 첫 번째 리밸런싱 단계)

6. 제외된 파티션 할당을 위해 컨슈머들은 다시 합류 요청함. 여기서 두 번째 리밸런싱이 트리거됨 (③ 두 번째 리밸런싱 단계)

7. 컨슈머 그룹의 리더는 제외된 파티션을 적절한 컨슈머에게 할당 (③ 두 번째 리밸런싱 단계)

재배치가 필요하지 않은 컨슈머들은 그대로 유지하여 다운타임 없이 동작. 현대 동작하고 있는 컨슈머에게 아무런 영향을 주지 않음

한 번에 일괄 작업하는 EAGER 방식 보다 COOPEPRATIVE 프로토콜 방식이 더 빠름. (211p 오타 발견)

6.5 정확히 한 번 컨슈머 동작

5.4 절에 있는 정확히 한 번 전송이 성공시 해당 레코드의 트랜잭션 성공을 표시하는 특수한 메시지를 추가했었음

그러므로 커슈머는 트랜잭션 코디네이터가 특수한 메시지를 표시한 레코드를 읽는다면 정확히 한 번만 읽을 수 있음

public class ExactlyOnceConsumer{

    public static void main(String[] args) {
        Properties props = new Properties();

        /* 정확히 한 번을 위한 설정 */
        props.setProperty(ConsimverConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    }
}

ISOLATION_LEVEL_CONFIG 설정을 추가하면 트랜잭션 컨슈머로 동작

트랜잭션 컨슈머는 트랜잭션 코디네이터와 통신하는 부분은 없으며 트랜잭션이 완료된 메시지만 읽을 수 있음

read_uncommitted는 모든 메시지를 읽을 수 있으며 read_committed는 트랜잭션이 완료된 메시지만 읽을 수 있음

한 건의 메시지를 전송했지만 트래잭션의 종료를 표시하기 위해 메시지 전송 후 커밋 또는 중단에 대한 표시를 남기는 트랜잭션 메시지를 추가해서 오프셋이 2이다.

(궁금한 점. 그렇다면 홀 수면 그 전에는 무조건 실패하는게 맞는 건지? 실패하여 1이고, 그 후에 메시지가 생성되어 성공햇다면 3이 될텐데 이는 실패로 볼 것인지?)

트랜잭션 메시지는 트랜잭션 성공여부만 확인하고 사용자에게 보여주지 않음

트랜잭션 컨슈머라고 해서 정확히 한 번만 가져오는 것은 아님

컨슈머의 경우 트랜잭션 프로듀서가 보낸 메시지만 가져올 수 있는지에 대해서만 옵션으로 선택할 수 있음

컨슈머는 트랜잭션 코디네이터와 통신하지 않으므로 정확하게 메시지를 한 번 가져오는지 보장할 수 없다? 뭔말이지 이게 코디네이터가 특수한 메시지를 표시한 레코드를 읽으면 보장하는 거 아닌가?

다른 싱크 저장소로 중복 저장될 수 있음

컨슈머의 동작까지 정확히 한 번 처리가 가능하려면 컨슘-메시지 처리-프로듀싱 동작 모두 하나의 트랜잭션으로 해결해야함