[Kafka] Consumer의 내부 동작 원리와 구현

* 컨슈머의 역할 : 오프셋 핵심

Consumer는 카프카에 저장된 메시지를 꺼내오는 역할을 하기 때문에, 컨슈머가 메시지를 어디까지 가져왔는지를 표시하는 것은 매우 중요하다.

- 코드 배포로 인해 컨슈머가 일시적으로 동작을 멈추고 재시작하는 경우

- 컨슈머가 구동 중인 서버에서 문제가 발생해 새로운 컨슈머가 기존 컨슈머의 역할을 대신하는 경우

기존 컨슈머의 마지막 메시지 위치부터 새로운 컨슈머가 메시지를 가져올 수 있어야만 장애로부터 빠르게 복구 가능

컨슈머 그룹은 자신의 오프셋 정보를 카프카에서 가장 안전한 저장소인 토픽에 저장하는데, _consumer_offsets 토픽에 컨슈머 그룹별로 오프셋 위치 정보가 기록된다.

[_consumer_offsets 토픽에 기록되는 내용]

: 컨슈머 그룹, 토픽, 파티션 등의 내용을 통합

_consumer_offsets 토픽에 기록된 정보를 이용해 컨슈머 그룹은 자신의 그룹이 속해 있는 컨슈머의 변경이 발생하는 경우 어느 위치까지 읽었는지를 추적할 수 있다.

offsets.topic.num.partitions : 기본값 50
offsets.topic.replication.factor : 기본값 3

 

Group Coordinator

: 컨슈머 그룹이 구독한 토픽의 파티션들과 그룹의 멤버들을 트래킹

컨슈머들은 하나의 컨슈머 그룹의 구성원으로 속하며, 컨슈머 그룹 내의 각 컨슈머들은 서로 자신의 정보를 공유하면서 하나의 공동체로 동작한다. 컨슈머 그룹 내 컨슈머들은 언제든지 자신이 속한 컨슈머 그룹에서 떠날 수 있으며 새로운 컨슈머가 합류할 수도 있다. 따라서 컨슈머 그룹은 이러한 변화를 인지하고 각 컨슈머들에게 작업을 균등하게 분배해야 한다. 컨슈머 그룹에서 각 컨슈머들에게 작업을 균등하게 분해하는 동작을 컨슈머 리밸런싱이라고 부르거나 컨슈머 리밸런싱이 일어났다고 표현한다.

[컨슈머 그룹 등록 과정]

1. 컨슈머는 컨슈머 설정값 중에서 bootstrap.brokers 리스트에 있는 브로커에게 컨슈머 클라이언트와 초기 커넥션을 연결하기 위한 요청을 보낸다.

2. 해당 요청을 받은 브로커는 그룹 코디네이터를 생성하고 컨슈머에게 응답을 보내는데, 컨슈머 그룹의 첫 번째 컨슈머가 등록될 때까지 아무 작업도 일어나지 않는다.

3. 그룹 코디네이터는 group.initial.rebalance.delay.ms의 시간 동안 컨슈머의 요청을 기다린다.

4. 컨슈머는 컨슈머 등록 요청을 그룹 코디네이터에게 보낸다. 이때 가장 먼저 요청을 보내는 컨슈머가 컨슈머 그룹의 리더가 된다.

5. 컨슈머 등록 요청을 받은 그룹 코디네이터는 해당 컨슈머 그룹이 구독하는 토픽 파티션 리스트 등 리더 컨슈머의 요청에 응답을 보낸다.

6. 리더 컨슈머는 정해진 컨슈머 파티션 할당 전략에 따라 그룹 내 컨슈머들에게 파티션을 할당한 뒤 그룹 코디네이터에게 전달한다.

7. 그룹 코디네이터는 해당 정보를 캐시하고 각 그룹 내 컨슈머들에게 성공을 알린다.

8. 각 컨슈머들은 각자 지정된 토픽 파티션으로부터 메시지들을 가져온다.

 

컨슈머 그룹은 그룹 코디네이터와 연결되어 관리를 받게 되고, 컨슈머들은 현재 자신들이 속한 컨슈머 그룹에서 빠져나갈 수도, 새롭게 합류할 수도 있다. 컨슈머 그룹의 변화는 컨슈머 코디네이터에게 컨슈머가 join 또는 leave  요청을 보냄으로써 자연스럽게 처리된다. 

Consumer Option	Value	Explanation
heartbeat.interval.ms	3000	기본값은 3000이며, 그룹 코디네이터와 하트비트 인터벌 시간이다. 해당시간은 session.timeout.ms보다 낮게 설정해야 하며, 1/3 수준이 적절하다
session.timeout.ms	10000	기본값은 10000이며, 어떤 컨슈머가 특정 시간 안에 하트비트를 받지 못하면 문제 발생했다고 판단해 컨슈머 그룹에서 해당 컨슈머는 제거되고 리밸런싱 동작이 일어난다.
max.poll.interval.ms	300000	기본값은 300000이며, 컨슈머는 주기적으로 poll()을 호출해 토픽으로부터 레코드들을 가져오는데, poll() 호출 후

그룹 코디네이터는 하트비트 옵션을 통해 컨슈머의 상태를 확인하며, 특정 컨슈머에 문제가 발생했다고 판단되면 컨슈머 리밸런싱 동작을 통해 컨슈머 그룹의 전체 균형을 다시 맞춘다. 단순하게 하트비트만으로도 컨슈머의 상태를 체크할 수 있지만 할당된 파티션에서 컨슈머가 정상적으로 메시지를 가져가고 있는지를 poll() 동작 여부를 통해 확인하는 등 카프카의 세심하고 상세한 설정을 할 수 있다.

Static Membership

하드웨어 점검이나 소프트웨어 업데이트 등의 이유로 관리자는 컨슈머 그룹 내의 컨슈머들을 하났기 순차적으로 재시작하고 싶은 경우가 있을 것이다. 하지만 하나의 컨슈머가 재시작될 때마다 전체 리밸런싱이 일어나며, 리밸런싱 작업이 일어나는 동안 컨슈머들은 일시 중지되기 때문에 매우 번거로운 일이 될 수 있다. 

컨슈머 리밸런싱?

일반적인 컨슈머 그룹 동작에서는 각 컨슈머를 식별하기 위해 엔티티 ID를 부여한다. 컨슈머 설정 변경이나 소프트웨어 업데이트로 인해 컨슈머가 재시작되면 컨슈머 그룹 내의 동일한 컨슈머임에도 새로운 컨슈머로 인식해 새로운 엔티티 ID가 부여되고 이로 인해 컨슈머 그룹의 리밸런싱이 발생하는 것이다. 이런 메커니즘은 대용량 메시지들을 처리하는 컨슈머 그룹에 적합하지 않다.

이를 해결하기 위해 스태틱 멤버십이라는 개념을 도입했는데, 스태틱 멤버십은 컨슈머 그룹 내에서 컨슈머가 재시작 등으로 그룹에서 나갔다가 다시 합류하더라도 리밸런싱이 일어나지 않게 한다. 컨슈머마다 인식할 수 있는 ID를 적용함으로써 다시 합류 시에도 그룹 코디네이터가 기존 구성원임을 인식할 수 있게 하는 것이다. 이 기능이 적용된 컨슈머는 그룹에서 떠날 때 그룹 코디네이터에게 알리지 않으므로 불필요한 리밸런싱도 발생하지 않는다.

스태틱 멤버십이 적용된 컨슈머의 경우, 

1. 컨슈머 그룹에서 떠날 때 그룹 코디네이터에게 알리지 않음

2. 다시 컨슈머 그룹에 합류 시에도 그룹 코디네이터는 ID 확인 후 기존 구성원임을 인지

총 2번의 불필요한 리밸런싱을 회피할 수 있다. 

스태틱 멤버십을 적용하려면 기본값이 null string인 group.instance.id만 설정해주면 된다. (Apache Kafka 2.3 이상) 이 옵션에는 그룹 코디네이터가 컨슈머를 식별할 수 있게끔 컨슈머 인스턴스 별로 고유한 값을 입력해야 한다. 스태틱 멤버십 기능을 적용한다면, session.timeout.ms를 기본값보다는 큰 값으로 조정해야 한다. 그 이유는 컨슈머를 재시작한 후 session.timeout.ms 값에 지정된 시간 동안 그룹 코디네이터가 하트비트를 받지 못한다면 강제로 리밸런싱이 일어나므로, 불필요한 리밸런싱 동작을 최소화하기 위한 스태틱 멤버십의 목적에 위배되기 때문이다. 예를 들어, 컨슈머 재시작시간이 총 2분 소요된다면 session.timeout.ms 값은 2분보다 큰 값으로 설정해야 불필요한 리밸런싱 동작을 사전에 방지할 수 있다.

컨슈머의 리밸런싱은 재시작되거나 그룹에서 떠나는 컨슈머만 대상으로 동작하는 것이 아니라 컨슈머 그룹 내 전체 컨슈머를 대상으로 동작함을 명심하자. 리밸런싱 과정에서 불가피하게 벌어지는 일시 중지 동작은 매우 부담이 크고 고비용 작업이기 때문에, 불필요한 리밸런싱을 최대한 줄여야 한다.

컨슈머 파티션 할당 전략

컨슈머의 파티셔너는 대상 토픽의 어느 파티션으로부터 레코드를 읽어올지를 결정한다.

컨슈머 그룹의 리더 컨슈머가 정해진 파티션 할당 전략에 따라 각 컨슈머와 대상 토픽의 파티션을 매칭시킨다. 파티션 할당 전략은 컨슈머 옵션의 partition.assignment.strategy로 표시하며, RangeAssignor / RoundRobinAssignor / StickyAssignor / CooperativeStickyAssignor의 총 4가지를 제공한다.

 

파티션 할당 전략

파티션 할당 전략	설명
RangeAssignor	파티션 할당 전략의 기본값으로서 토픽별로 할당 전략을 사용함. 동일한 키를 이용하는 2개 이상의 토픽을 컨슘할 때 유용
RoundRobinAssignor	사용 가능한 파티션과 컨슈머들을 라운드 로빈으로 할당하여 균등한 분배를 가능하게 한다.
StickyPartitionAssignor	컨슈머가 컨슘하고 있는 파티션을 계속 유지할 수 있다
CooperativeStickyAssignor	스티키 방식과 유사하지만 전체 일시 정지가 아닌 연속적인 재조정 방식

RangeAssignor

먼저 구독하는 토픽에 대한 파티션을 순서대로 나열한 후 컨슈머를 순서대로 정렬한다. 그런 다음 각 컨슈머가 몇 개의 파티션을 할당해야 하는 지 전체 파티션 수를 컨슈머 수로 나눈다. 이 때 컨슈머 수와 파티션 수가 일치하면 균등하게 할당될 수 있으나 균등하게 나눠지지 않는 경우 앞쪽의 컨슈머들이 추가 파티션을 할당받게 된다.

- 전체 (토픽당) 파티션 수 / 전체 컨슈머 수 = 3/2 = 1.5 : 컨슈머당 최소 하나의 파티션을 가져야 함

균등하게 나눠지지 않으므로 먼저 정렬된 컨슈머인 컨슈머1에 남은 파티션을 추가 할당한다.결국 컨슈머 1은 4개의 파티션을, 컨슈머 2는 2개의 파티션을 담당하게 된다.

이러한 파티션 할당 전략을 사용하는 이유는, 동일한 메시지 키를 사용하고 하나의 컨슈머 그룹이 동일한 파티션 수를 가진 2개 이상의 토픽을 컨슘할 때 유용하다. 동일한 키값을 갖고 잇는 두 토픽의 파티션을 하나의 컨슈머가 컨슘하기 때문이다.

 

RoundRobinAssignor

라운드 로빈 파티션 할당 전략은 파티션 할당 전략 중 가장 간단한 할당 방식이다. 컨슘해야 하는 모든 파티션과 컨슈머 그룹 내 모든 컨슈머를 나열한 후 라운드 로빈으로 하나씩 파티션과 컨슈머를 할당하는 방식이다.

라운드 로빈 할당 전략은 먼저 구독 대상 토픽의 전체 파티션을 나열한다. 이후 전체 컨슈머들도 나열한 다음 하나씩 라운드 로빈 방식으로 일대일 매핑한다.

파티션	매핑된 컨슈머
토픽1-파티션0	컨슈머1
토픽1-파티션1	컨슈머2
토픽1-파티션2	컨슈머1
토픽2-파티션0	컨슈머2
토픽2-파티션1	컨슈머1
토픽2-파티션2	컨슈머2

표와 같이 하나씩 번갈아가며 컨슈머를 할당하므로, RangeAssignor보다 RoundRobinAssignor가 더 균등히 파티션을 할당한다고 할 수 있다.

StickyAssignor

컨슈머 그룹의 리밸런싱 동작으로 인해 파티션이 재할당된다면 어떤 상황이 벌어질까? 레인지 파티션 할당 전략과 라운드 로빈 할당 전략 모두 파티션 재할당 작업이 발생하면 기존에 매핑됐던 파티션과 동일한 컨슈머가 다시 매핑된다고 보장할 수 없다. 이런 재할당 작업이 발생하더라도 기존 매핑되었던 파티션과 컨슈머를 최대한 유지하려고 하는 전략이 바로 스티키 파티션 할당 전략이다.

스티키 파티션 할당 전략은 두 가지 목적으로 컨슈머에 파티션을 할당한다. 첫번째는 가능한 한 균형 잡힌 파티션을 할당하고, 두번째로 재할당이 발생하면 되도록 기존의 할당된 파티션 정보를 보장하는 것이다. 두 가지 목적 중 첫번째에 우선순위가 있으며, 최대한 컨슈머를 균등하게 분배하는 것을 우선하므로 일부 파티션은 기존 컨슈머와 매핑을 유지하지 못하고 새로운 컨슈머와 연결될 수도 있다.

스티키 파티션 할당 전략에서도 최초의 배치 전략은 라운드 로빈 할당 전략과 매우 흡사하다. 하지만 리밸런싱이 일어났을 때 큰 차이가 있다. 다음의 상황을 가정하자.

[라운드 로빈 방식]

1. 컨슈머2가 컨슈머 그룹에서 떠남

2. 리밸런싱 동작이 일어남

3. 모든 파티션을 순서대로 배치

4. 모든 컨슈머를 순서대로 배치

5. 라운드 로빈 파티션 할당 전략에 맞춰 하나씩 매핑

파티션	매핑된 컨슈머
토픽1-파티션0	컨슈머1
토픽1-파티션1	컨슈머3
토픽2-파티션0	컨슈머1
토픽2-파티션1	컨슈머3
토픽3-파티션0	컨슈머1
토픽3-파티션1	컨슈머3
토픽4-파티션0	컨슈머1
토픽4-파티션1	컨슈머3

파티션 연결이 끊기고 컨슈머 그룹의 리밸런싱으로 인해 새로운 파티션이 할당하는 일이 발생한다.

[스티키 파티션 할당 전략]

파티션	매핑된 컨슈머
토픽1-파티션0	컨슈머1
토픽1-파티션1	컨슈머3
토픽2-파티션0	컨슈머3
토픽2-파티션1	컨슈머1
토픽3-파티션0	컨슈머1
토픽3-파티션1	컨슈머3
토픽4-파티션0	컨슈머1
토픽4-파티션1	컨슈머3

컨슈머2에 할당된 파티션들만 컨슈머1과 컨슈머3에 새로 각각 할당된 것을 알 수 있다.

스티키 파티션 할당 전략이 이상적으로 동작하는 이유는 다음과 같은 규칙에 따라서 재할당 동작을 수행하기 때문이다.

- 컨슈머들의 최대 할당된 파티션의 수 차이는 1

- 기존에 존재하는 파티션 할당은 최대한 유지

- 재할당 동작 시 유효하지 않은 모든 파티션 할당은 제거

- 할당되지 않은 파티션들은 균형을 맞추는 방법으로 컨슈머들에 할당

스티키 파티션 할당 전략은 최대한 컨슈머들의 균형을 맞추고 기존 컨슈머에 할당된 파티션을 최대한 유지함으로써 컨슈머에 새로 할당하는 파티션 수를 최소화한다. 이렇게 최소한의 움직임으로 컨슈머를 할당할 수 있으므로 라운드 로빈 전략보다 효율적이다.

 

CooperativeStickyAssignor

리밸런싱이 일어나도 기존의 컨슈머와 파티션 매핑은 유지하고 최소한의 파티션만 컨슈머와 매핑한다. 다만 협력적 스티키 파티션 할당 전략과 한 가지 차이점이 있는데, 그건 바로 컨슈머 그룹 내부의 리밸런싱 동작이 한층 더 고도화됐다는 점이다.

EAGER 프로토콜(감지 -> 중지 -> 재시작)

컨슈머 리밸런싱 동작시 컨슈머에 할당된 모든 파티션을 항상 취소한다. 이 프로토콜을 통해 컨슈머와 파티션 재할당이 가능했고 한 번의 리밸런싱 동작으로 모든 컨슈머와 파티션 매핑이 가능해졌다. 

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

: 하나의 컨슈머 그룹 내에서는 둘 이상의 컨슈머가 동일한 파티션을 소유할 수 없다.

2. 컨슈머들의 다운타임

: 리밸런싱에서 모든 파티션 할당을 취소  -> 리소스를 많이 사용하는 컨슈머 그룹에서는 큰 문제 발생

리밸런싱 전체 동작 과정

1. 감지 단계 : 컨슈머가 다운됨을 감지

2. 중지 단계 : 컨슈머에게 할당된 모든 파티션을 제거 -> 컨슈머에게 할당된 파티션이 없으므로 컨슈머의 다운타임이 시작

그러나, 컨슈머의 동작과 프로듀서의 동작은 완벽하게 분리되어 있으므로, 컨슈머가 리밸런싱하는 도중에도 프로듀서는 해당 토픽을 타깃으로 메시지를 지속적으로 전송한다. 결국 다운타임 동안 LAG가 급격히 증가한다.

3. 재시작 단계 : 구독한 파티션이 컨슈머들에게 재할당된다. 컨슈머들은 각자 할당받은 파티션에서 메시지들을 컨슘하기 시작하고 비로소 컨슈머의 다운타임이 종료된다. 앞서 설명한 스태틱 멤버십의 기능을 통해 불필요한 리밸런싱 동작이 일어나지 않게 할 수 있으나, 불가피하게 리밸런싱이 일어난다면 여전히 컨슈머의 다운타임은 발생한다. 

이를 해결하기 위해, 협력적 스티키 할당 전략이 등장한다.

COOPERATIVE 프로토콜

되도록 동작 중인 컨슈머들에게 영향을 주지 않는 상태에서 몇 차례에 걸쳐 리밸런싱이 이뤄진다. 협력적 스티키 파티션 할당 전략은 안전하게 파티션의 소유권을 이동하기 위해 리밸런싱 작업이 수차례에 걸쳐 진행하는 것도 나쁘지 않다는 아이디어에서 출발한다.

1. 컨슈머 그룹에 peter-kafka01이 합류하면서 리밸런싱 트리거

2. 컨슈머 그룹 내 컨슈머들은 그룹 합류 요청과 자신들이 컨슘하는 토픽의 파티션 정보를 그룹 코디네이터로 전송

3. 그룹 코디네이터는 해당 정보를 조합해 컨슈머 그룹의 리더에게 전송

4. 컨슈머 그룹의 리더는 현재 컨슈머들이 소유한 파티션 정보를 활용해 제외해야 할 파티션 정보를 담은 새로운 파티션 할당 정보를 컨슈머 그룹 멤버들에게 전달 

5. 새로운 파티션 할당 정보를 받은 컨슈머 그룹 멤버들은 현재 파티션 할당 전략과 차이를 비교해보고 필요 없는 파티션을 골라 제외한다. 이전의 파티션 할당 정보와 새로운 파티션 할당 정보가 동일한 파티션에 대해서는 어떤 작업도 수행할 필요가 없다.

6. 제외된 파티션 할당을 위해 컨슈머들은 다시 합류 요청 -> 두 번째 리밸런싱 트리거

7. 컨슈머 그룹의 리더는 제외된 파티션을 적절한 컨슈머에게 할당

협력적 스티키 파티션 할당 전략에서 중요한 점은 파티션 재배치가 필요하지 않은 컨슈머들은 다운타임 없이 계속 동작하며, 한 번이 아니라 두 차례의 리밸런싱이 일어났다는 점이다.  리밸런싱 동작이 한 번에 완료되지 않았음에도 현재 동작하고 있는 컨슈머에게는 아무런 영향을 주지 않았다.

컨슈머의 영향을 최소화하기 위해 점진적으로 리밸런싱이 진행되더라고, 컨슈머의 확장 및 축소, 롤링 재시작 등의 작업 진행 시 컨슈머 리밸런싱에 대한 부담을 줄일 수 있다.

정확히 한 번 컨슈머 동작

컨슈머는 트랜잭션 코디에니어가 표시한 특수한 메시지의 레코드만 읽는다면 정확히 한 번 읽을 수 있다.

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apahce.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Arrays;
import java.util.Properties;

public class ExactlyOnceConsumer {
	public static void main(String[] args) {
    	String bootstrapServers = "peter-kafka01.foo.bar:9092"
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    	props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
bootstrapServers);
    	props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "peter-consumer-01");
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("peter-test05"));
        
        try {
        	while (true) {
            	ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records){
                	System.out.printf("Topic: %s, Partitiion: %s, Offset: %d, Key: %s,
Value: %s\n", record.topic(), record.partition(), record.offset(), record.key(),
record.value());
                }
                consumer.commitAsync();
            }
        } catch (Exception e) {
        	e.printStackTrace();
        } finally {
        	consumer.close()
        }
    }
}

본 코드는 트랜잭션 컨슈머의 예제 코드이며, 일반 컨슈머 코드에서 ISOLATION_LEVEL_CONFIG라는 설정만 추가하면 트랜잭션 컨슈머로 동작한다. 트랜잭션 컨슈머는 트랜잭션 코디네이터와 통신하느 부분은 없으며 트랜잭션이 완료된 메시지만 읽을 수 있다. 기본값은 read_uncommitted로서 모든 메시지를 읽을 수 있다는 뜻이며, read_committed로 옵션값을 변경하면 트랜잭션이 완료된 메시지만 읽을 수 있게 된다. 

하지만 트랜잭션 컨슈머라고 해서 정확히 한 번만 가져올 수 있는 것은 아니다. 프로듀서의 경우 트랜잭션 코디네이터와 통신하면서 해당 트랜잭션이 정확하게 처리되는 것을 보장했지만, 컨슈머의 경우  프로듀서가 보낸 메시지만 가져올 수 있는지에 대해서만 옵션으로 선택할 수 있다. 컨슈머는 트랜잭션 코디네이터와 통신하는 부분이 없으므로 정확하게 메시지를 한 번 가져오는지는 보장할 수 없다. 또한 컨슈머에 의해 컨슘된 메시지가 다른 싱크 저장소로 중복 저장될 수 있다.

만일 컨슈머가 정확히 한 번 메시지를 가져왔다고 가정하더라도, 컨슈머가 가져온 메시지를 다른 애플리케이션에 저장하는 과정에서 중복 처리되는 경우가 있다. 카프카 클라이언트인 컨슈머는 다른 싱크 저장소로 메시지들이 중복 저장되는 결과를 알 수 없으므로 정확히 한 번 저장을 보장할 수 없다.