Chapter 06. 컨슈머의 내부 동작 원리와 구현

[MinJunKweon]

[minkukjo]

끄적끄적

• 컨슈머 그룹이 생성되면 그룹 코디네이터가 생기게 됨
• 그룹 코디네이터와 컨슈머는 서로 하트비트를 주고받으면서 컨슈머들이 잘 살아 있는지, 새로운 컨슈머가 join하거나 기존 컨슈머가 leave되는지를 관찰함
• 물론 당연히 컨슈머가 새로 join하거나 leave되는 경우 컨슈머 리밸런싱이 발생하므로 조심해야함
• 스태틱 멤버십으로 지정된 컨슈머는 빠지거나 합류하더라도 리밸런싱이 일어나지 않는다.
• 물론 아예 안일어나는건 아니고, session.timeout.ms 지정 시간을 넘어가지 않으면 그 시간 동안 컨슈머가 잠시 제외되더라도 리밸런싱이 발생하지 않는다. ( 그 시간 이후면 리밸런싱 발생 )
• 컨슈머 파티션은 총 네 개의 전략을 사용함. 레인지, 라운드 로빈, 스티키, 협력적 스티키

• 공식문서를 살펴보니 3버전부터는 기본적으로 컨슈머 파티션 전략은 레인지와 협력적 스티키 두 개가 디폴트 설정으로 잡혀있다고 합니다. 디폴트는 레인지고, 레인지를 리스트에서 제거하면 협력적 스티키가 된다고 하네요. 공식 도큐먼트

• exactly once는 트랜잭션 코디네이터와 통신하는 부분은 따로 없고, 트랜잭션이 완료된 메시지만 읽을 수 있다. 기본적으로 read_uncommitted이고 추후 read_committed로 옵션을 변경하면 트랜잭션이 완료된 메시지만 읽을 수 있다.

• 컨슈머 자체에는 트랜잭션을 보장하는 코디네이터와의 통신이 없으므로 컨슈머 자체만의 트랜잭션 처리로는 중복이 발생할 여지가 존재한다.

• 따라서 sendOffsetsToTransaction 메소드를 이용해, 컨슈머 그룹의 오프셋 커밋을 트랜잭션에 포함시켜서, 컨슈머-메시지 처리-프로듀싱 동작이 모두 하나의 트랜잭션으로 동작하게 되야한다.

[MinJunKweon]

궁금한 내용

• 201p 컨슈머1에 4개 파티션이 할당되었다고 하는데, 같은 컨슈머 그룹 ID를 사용하는 컨슈머면 모든 토픽에서 컨슈머 엔티티 ID가 같은 것인지?
• 1개의 컨슈머 그룹으로 2개 이상의 토픽을 컨슈밍해야하는 경우가 있을까요?
• sendOffsetsToTransaction() 메소드는 프로듀서의 메소드인데, 그러면 프로듀서가 컨슈밍까지 완료하는 것 아닌가?
  • 그러면 카프카에 굳이 메시지를 전송해야함? 이미 컨슈밍 된 거 아님?
• 카프카 컨슈머 commit에도 브로커는 ack를 보낼까? YES. 보냅니다. 클라이언트가 기다려요.
  • TMI) 코드 보니까 카프카 클라이언트들은 응답을 기다릴 때 Selector를 쓰네요
• 그렇다면 이미 커밋된 메시지를 컨슈머가 중복으로 처리되는 경우가 있을까?
• partition.assignment.strategy가 여러 개면 어떻게 동작하는지? 두 개를 같이 쓸 수 있는걸까요?

---

컨슈머 오프셋

• 컨슈머가 브로커에 저장된 메시지를 어디까지 가져왔는지 표시하는 역할
• 컨슈머가 일시적으로 동작을 멈췄다가 재시작하는 경우 등 어느 메시지까지 컨슈밍이 되었는지 확인할 필요가 있음
• 오프셋 정보는 컨슈머 그룹별로 관리되며, __consumer_offsets 토픽에 저장됨
• __consumer_offsets의 파티션 수, 리플리케이션 팩터는 server.properties 파일에서 변경 가능
  • offsets.topic.num.partitions : 기본값 50
  • offsets.topic.replication.factor : 기본값 3

그룹 코디네이터

• 컨슈머 그룹 내의 컨슈머들이 떠나고 합류하는 것을 적절히 분배하는 것을 ‘컨슈머 리밸런싱’이라고 부름
• 안정적인 컨슈머 리밸런싱을 위해 별도의 그룹 코디네이터를 사용함
• 컨슈머 그룹이 최초 연결 요청을 보내면 브로커들 중 하나에 그룹 코디네이터가 생성됨
• 컨슈머 멤버 변경, 구독하는 토픽 파티션 변경 등 감지하여 컨슈머에게 공지함

그룹 코디네이터 동작 순서

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

• 이후 컨슈머가 합류하거나 떠날 때, 컨슈머는 코디네이터에 Join 혹은 Leave 요청을 보내서 처리함

만약 컨슈머가 Leave 요청을 보내지 않고 종료된다면 코디네이터는 어떻게 감지할 수 있는가?

• 컨슈머들은 코디네이터에 주기적으로 하트비트(heartbeat)를 전송해서 살아있음을 알림
• 만약 컨슈머로부터 하트비트가 오지 않으면 문제가 발생한 것으로 간주해서 리밸런싱함
• 하트비트 관련된 옵션은 아래와 같음

컨슈머 옵션	설명	기본값
heartbeat.interval.ms	그룹 코디네이터와 하트비트 인터벌 시간 해당 시간은 session.timeout.ms보다 낮게 설정해야 함. 일반적으로 1/3 수준	3000 (3초)
session.timeout.ms	컨슈머와 브로커사이의 session timeout 시간 이 시간이 지나면 해당 컨슈머는 종료되거나 장애가 발생한 것으로 간주되어 리밸런싱이 일어남	10000 (10초)
max.poll.interval.ms	컨슈머 poll() 호출 인터벌 시간 컨슈머가 poll()을 호출 후 최대 설정시간만큼 지나도 다시 poll()을 호출하지 않으면 문제가 있는 것으로 판단	300000 (5분)

스태틱 멤버십

• 여러 대의 컨슈머를 순차적으로 재시작해야하는 경우가 있을 수 있음 (ex. 점검, 배포 등)
• 컨슈머가 순차적으로 재시작 될 때마다 리밸런싱이 일어나게되고, 리밸런싱 도중에는 정상동작 중인 컨슈머들이 컨슈밍을 못함
• 스태틱 멤버십은 컨슈머 그룹 내에 컨슈머가 재시작 되어도 리밸런싱이 일어나지 않게 막음
• 컨슈머마다 인식할 수 있는 ID를 적용하여 다시 합류하더라도 리밸런싱을 하지 않도록함
• 스태틱 멤버십이 적용된 컨슈머는 떠날 때 코디네이터에 알리지 않음(리밸런싱 방지)
• 컨슈머 옵션 중 group.instance.id를 통해 스태틱 멤버십을 적용할 수 있음
• 스태틱 멤버십은 session.timeout.ms를 재시작 하는 시간보다 길게 설정해야함
  • 재시작 후 하트비트를 전송하기까지 오래걸리면 리밸런싱이 일어나게 되기 때문

컨슈머 파티션 할당 전략

파티션 할당 전략	설명
Range 파티션 할당 전략	토픽별로 할당 전략을 사용함. 동일한 키를 이용하는 2개 이상의 토픽을 컨슘할 때 유용함 (기본값)
Round-Robin 파티션 할당 전략	사용 가능한 파티션과 컨슈머들을 라운드 로빈으로 할당함.
Sticky 파티션 할당 전략	컨슈머가 컨슘하고 있는 파티션을 계속 유지할 수 있음
Cooperative Sticky 파티션 할당 전략	스티키 방식과 유사하지만, 전체 일시 정지가 아닌 연속적인 재조정 방식 (3.0 버전 이상 기본값)

레인지 파티션 할당 전략 (기본값)

• 각 토픽별로 파티션을 순서대로 정렬해서 전체 파티션 수를 컨슈머 수로 나누어 할당함
• 딱 떨어지지 않는 경우 앞쪽의 컨슈머들이 추가 파티션을 할당 받음
• 예를 들어, 파티션 3개 - 컨슈머 2개라면 (컨슈머1 - 파티션0, 파티션1), (컨슈머2 - 파티션2) 이렇게 할당됨
• 여러 개의 토픽을 사용할 때 앞쪽에 있는 컨슈머가 많은 파티션을 할당받게 되므로 불균형하다는 단점이 있음

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

• 모든 토픽의 모든 파티션을 정렬해서 컨슈머들에 균등하게 하나씩 분배함

스티키 파티션 할당 전략

• 앞서 2개의 전략은 리밸런싱이 일어나는 경우 다시 같은 파티션을 할당하리라는 보장이 없음
• 리밸런싱이 일어나더라도 기존에 매핑된 파티션과 컨슈머를 최대한 유지하려고 하는 전략
• 하지만, 최대한 균형 잡힌 파티션 할당을 하는 것이 우선순위가 높기 때문에 항상 같은 파티션을 컨슈밍한다는 보장을 하지않음
• 최초 접근 시 파티션 할당 전략은 라운드 로빈 전략처럼 모든 파티션을 각 컨슈머에 하나씩 분배함
• 리밸런싱이 일어나면 떠난 컨슈머의 파티션만 나머지 컨슈머들이 분배받는 방식(기존 컨슈머들의 파티션은 변경없음)

협력적 스티키(Cooperative Sticky) 파티션 할당 전략

• 컨슈머 리밸런싱엔 내부적으로 EAGER 프로토콜을 사용하는데, 이는 컨슈머에 할당된 모든 파티션을 항상 취소함
  • 컨슈머들의 파티션 소유권을 변경하는 과정에서 잠시 컨슈머가 아무 컨슈머에도 할당되면 안되어야하기 때문
  • 여러 파티션들에 대해 소유권 변경하는 과정이 동시에 이뤄져야하므로 로직 단순화를 위해
• 파티션 소유권 변경과정에서 컨슈머가 동작을 하지 못하기 때문에 문제가 발생함(LAG이 급격히 증가함)
• 그렇기 때문에 EAGER 프로토콜 대신 COOPERATIVE 프로토콜을 사용함

협력적 프로토콜 방식

1. 컨슈머가 새로 합류하면 리밸런싱이 트리거됨
2. 컨슈머 그룹 내의 컨슈머들은 합류 요청과 자신들이 컨슘하는 토픽의 파티션 정보를 코디네이터에 전송함
3. 코디네이터가 정보를 조합해 리더 컨슈머에 전송함
4. 리더 컨슈머가 제외해야할 파티션 정보를 담아서 그룹 멤버들에게 전달함
5. 그룹 멤버들은 필요없는 파티션을 골라 제외함. 만약 제외할 게 없으면 아무 작업을 하지 않음
6. 제외된 파티션 할당을 하려고 컨슈머들이 다시 합류 요청을 보냄
7. 리더 컨슈머는 제외된 파티션을 적절히 컨슈머들에 할당함

정확히 한 번(Exactly-once) 컨슈머 동작

• 컨슈머는 isolation.level 옵션에 read_committed 값을 주면 트랜잭션이 완료된 메시지만 읽을 수 있게 됨
• 하지만, 컨슈머의 이 옵션은 트랜잭션 프로듀서가 보낸 메시지만 가져올 수 있는지만 보장함
  • 컨슈머는 트랜잭션 코디네이터와 통신을 하지 않기 때문에 정확히 한 번만 메시지를 가져오는지 보장할 수 없음
  • 컨슈머에 의해 컨슘된 메시지가 다른 싱크 저장소로 중복 저장될 수 있음
  • 컨슈머가 정확하게 한 번 메시지를 가져오더라도 다른 애플리케이션에 저장하는 과정에서 중복 처리 될 수 있음
• 컨슈머의 동작까지 정확히 한 번 처리가 가능하려면 ‘Consume-Processing-Producing’이 모두 하나의 트랜잭션으로 처리되어야함
  • sendOffsetsToTransaction() 메소드를 사용하면 컨슈머 그룹의 오프셋까지 트랜잭션에 추가할 수 있음
• 컨슈머 애플리케이션 레벨에서 정확히 한 번을 지원하는 경우가 있으니 확인바람!

[LOG-INFO]

끄적끄적

• 컨슈머 그룹별 파티션 오프셋 정보는 __consumer_offsets 토픽에 저장된다
  • 여기서 저장되는 오프셋 정보는 마지막까지 읽은 위치가 아니라 다음으로 읽어야 할 위치
• 안정적인 컨슈머 그룹 관리를 위해 그룹 코디네이터가 존재함
  • 1 컨슈머 그룹 : 1 그룹 코디네이터
  • 클러스터 내 브로커 중 하나 지정
  • 컨슈머 그룹이 구독한 토픽의 파티션들과 컨슈머 그룹의 멤버들을 트래킹함
  • 둘 중 하나에 변화가 생기면 컨슈머 리밸런싱을 발생시킴
• 한 컨슈머 그룹에는 리더 컨슈머가 존재
  • 컨슈머 등록 요청을 가장 먼저 한 컨슈머가 리더 컨슈머가 됨
  • 정해진 파티션 할당 전략에 따라 그룹 내 컨슈머들에게 파티션을 할당(assign)한 뒤 그룹 코디네이터에게 전달
    • partition.assignment.strategy
      • RangeAssignor (default)
      • RoundRobinAssignor
      • StickyAssignor
      • CooperativeStickyAssignor
• 컨슈머 그룹의 변화
  • 컨슈머의 join, leave 요청
  • 컨슈머의 heartbeat
    • heartbeat.interval.ms
    • session.timeout.ms
  • 컨슈머의 poll 요청 간격
    • max.poll.interval.ms
    • (+ max.poll.records)
• 스태틱 멤버십
  • 특정 컨슈머에 고정된 group.instance.id를 부여하면 스태틱 멤버로 지정됨
  • 그러면 해당 컨슈머가 잠깐 다운되어도 바로 컨슈머 리밸런싱이 일어나지 않음
    • 기본적으로는 배포할 때마다 컨슈머 리밸런싱 일어나는데, 이것을 방지할 수 있을 듯
    • 단, session.timeout.ms 안에는 다시 실행되어야 함
• Exactly Once
  • 컨슈머는 트랜잭션 코디네이터가 특수한 메시지(컨트롤 메시지)를 표시한 레코드만 읽는다면 정확히 한 번 읽을 수 있다
  • isolation.level: read_committed

궁금한 점

• java kafka-client 기준, 한 컨슈머에 여러 partition이 할당되면, 한 스레드로 모든 파티션을 컨슈밍할까? 파티션 개수만큼의 스레드로 각각 컨슈밍할까?

좋은 글 추천

• https://www.confluent.io/blog/kafka-consumer-multi-threaded-messaging/
  • consuming, processing, commit은 한 thread에서 이루어 진다 (poll())
    • auto commit이 별도 스레드에서 이루어질 것이라 예상했지만, 아니다. 같은 스레드에서 이루어진다.
  • auto.commit.interval.ms는 커밋 최소 시간일 뿐, 무조건 보장하지 않는다. 더 넘어갈 수 있다.

[LOG-INFO]

Understanding Kafka partition assignment strategies and how to write your own custom assignor
위 문서를 보면 partition.assignment.strategy가 list인 이유를 알 수 있네여.
컨슈머 그룹 내의 모든 컨슈머들이 같은 프로토콜을 지원한다는 보장이 없어서 여러 개 지정하는 듯
HTTP의 Accept 헤더 같은 느낌

[LOG-INFO]

worker thread 병렬처리(여러 파티션 consuming 병렬처리 말고) 중 늦은 메시지가 먼저 처리되면?