카프카 컨슈머 프로토콜 비교하기 ClassicKafkaConsumer vs AsyncKafkaConsumer

카프카를 살펴보다가 ConsumerDelegate를 구현하고 있는 두가지 클래스를 볼 수 있었습니다. ClassicKafkaConsumer 와 AsyncKafkaConsumer인데요 컨슈머는 왜 2가지 방식으로 존재하는지 궁금했고 흥미로워서 이에대해서 정리를 해보고자 합니다.

우선 클래스명을 보고 ClassicKafkaConsumer 가 좀 더 오래된 컨슈머겠구나 예상을 하고 들여다 봤습니다. 해당 클래스에는 주석부터 classic group protocol 을 사용하는 클라이언트라고 되어 있었습니다. 그럼 classic group protocol이란 무엇인가..?

찾은 문서를 기반으로 설명하면 classic protocol/classic group protocol 으로 부르고 다음과 같은 특징이 있다고 합니다.

중앙 조정자(Group Coordinator) + 전원 동기화 방식

• 리밸런스가 시작되면 그룹 전체가 join/ synk barrier 에서 멈추고 기다려야 합니다.
• 한 멤버만 느리거나(join 요청 등) 장애가 발생해도 그룹 전체가 영향을 받습니다.

좀 더 자세히 살펴보면

Classic Protocol의 문제점 (KIP-848 문서 기준)

• 전원 동기화로 인한 높은 비용
  • 모든 멤버가 JoinGroup → SyncGroup을 다시 해야 하므로 큰 소비자 그룹일수록 리밸런스 비용이 매우 커짐
  • 한 멤버만 비정상이어도 그룹 전체가 리밸런스 지연됨
• 리밸런스에서 클라이언트가 많은 부분을 담당하게 됨.
  • 클라이언트 버전/로그가 없으면 운영자 입장에서 원인 파악이 거의 불가능
• 클라이언트 버전/로그가 없으면 운영자 입장에서 원인 파악이 거의 불가능
  • 여전히 JoinGroup / SyncGroup barrier는 존재함
  • 부분적 revocation은 가능하지만 그룹 전체에 영향을 주는 구조는 그대로

Classic Protocol 동작

✔ Join → Sync 두 단계 프로토콜

• 모든 멤버가 Join → Sync 두 단계 프로토콜
• 두 단계에서 전체 멤버를 기다려야 한다는 점을 큰 문제

✔ 클라이언트 assignor가 과하게 많은 역할을 맡고 있음

• 모든 멤버 metadata를 리더 컨슈머가 받아야 하고
• 실제 할당 계산도 클라이언트가 수행
• 서버에서 관찰/제어가 어려움

JoinGroup 단계에서는 모든 멤버가 Join을 시도해야 하고 코디네이터 브로커는 일정 시간 동안 모든 멤버의 JoinGroup을 기다리게 됩니다. 멈추는 이유에 대해서는 코디네이터가 그룹 리더를 정하고 모든 멤버의 정보를 합쳐서 리더 컨슈머에게 알고리즘 실행을 명령하기 위함입니다.

(1) RebalanceInProgress 발생
(2) 모든 멤버 작업 중단
(3) 모든 멤버 JoinGroup 전송 → 모든 멤버 때문에 기다림 ← Stop-the-World
(4) JoinGroupResponse: 리더/멤버 정보 확정
(5) 리더가 SyncGroup 전송
(6) 코디네이터가 할당 전송
(7) 모든 멤버가 새 파티션으로 작업 재개

CooperativeStickyAssignor 같은 향상된 rebalance를 사용해도 파티션 이동은 부분적으로만 일어나지만 JoinGroup 전체 베리어는 그대로 유지되는 문제가 있습니다.

 

그럼 새로운 프로토콜에서는 어떻게 변경이 됐을까요?

 

새로운 프로토콜

 

우선 저도 이전 프로토콜에서 가장 이해가 안됐던 부분이 어차피 코디네이터가 컨슈머 정보를 갖고있는데 왜 분배를 리더한테 맡기는거지? 라는 의문이 들었었는데요. 해당 부분을 이제는 코디네이터에서 하도록 변경되었습니다.

 

기존에는 리더 컨슈머가 assignor 실행 → SyncGroup으로 멤버에 브로드캐스트 하던 방식을 이제는 코디네이터 브로커가 직접 계산하고 컨슈머는 단순히 본인이 가진 파티션 목록을 받게 됐습니다.

그리고 이제는 JoinGroup이 제거 되어 동시에 합류를 바라는것이 아닌 각 멤버가 독립적으로 코디네이터에게 상태를 업데이트하게 되고 전체를 기다릴 필요 없이 상대를 바로 반영하게 됩니다.

1) 어떤 이벤트로 그룹 epoch 증가
2) Broker에서 새로운 desired assignment 계산
3) Broker가 멤버에게 목표 상태 전달
4) 멤버는 현재 상태와 비교하여:
    - 내려놓아야 하는 파티션만 revoke
    - 새로 가져야 하는 파티션만 추가
5) 멤버는 조정이 끝나면 ack
6) Stable 상태 도달

 

 

 

그 외에도 poll 과정에서 브로커에 fetch 요청을 보낼 때 기존에는 블락킹 방식으로 요청을 했으나 새로운 프로토콜에서는 논블락킹 방식으로 브로커에 fetch 요청을 보내는 등에 변경사항이 있었습니다. 자세한 내용은 아래 문서를 참고하시면 도움될것 같습니다!

 

 

 

KIP-848