MinJunKweon
commented
2 years ago 궁금한 내용
- 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: 기본값 50offsets.topic.replication.factor: 기본값 3
그룹 코디네이터
- 컨슈머 그룹 내의 컨슈머들이 떠나고 합류하는 것을 적절히 분배하는 것을 ‘컨슈머 리밸런싱’이라고 부름
- 안정적인 컨슈머 리밸런싱을 위해 별도의 그룹 코디네이터를 사용함
- 컨슈머 그룹이 최초 연결 요청을 보내면 브로커들 중 하나에 그룹 코디네이터가 생성됨
- 컨슈머 멤버 변경, 구독하는 토픽 파티션 변경 등 감지하여 컨슈머에게 공지함
그룹 코디네이터 동작 순서
- 컨슈머는 컨슈머 설정값 중에서
bootstrap.brokers리스트에 있는 브로커에게 컨슈머 클라이언트와 초기 커넥션을 연결하기 위한 요청을 보냄 - 해당 요청을 받은 브로커는 그룹 코디네이터를 생성하고, 컨슈머에게 응답을 보냄. 첫번째 컨슈머가 등록될 때까지 대기
- 그룹 코디네이터는
group.initial.rebalance.delay.ms의 시간 동안 컨슈머의 요청을 대기 - 컨슈머 등록 요청을 그룹 코디네이터에 전송. 가장 먼저 요청을 보내는 컨슈머가 리더 컨슈머
- 컨슈머 등록 요청을 받은 그룹 코디네이터는 해당 컨슈머 그룹이 구독하는 토픽 파티션 리스트 등을 리더 컨슈머에 응답
- 리더 컨슈머는 정해진 컨슈머 파티션 할당 전략에 다라 그룹 내 컨슈머들에게 파티션을 할당한 뒤 그룹 코디네이터에게 전달
- 그룹 코디네이터는 해당 정보를 캐시하고 각 그룹 내 컨슈머들에게 성공을 알림
- 각 컨슈머들은 각자 지정된 토픽 파티션으로부터 메시지 컨슈밍
- 이후 컨슈머가 합류하거나 떠날 때, 컨슈머는 코디네이터에 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 프로토콜을 사용함
협력적 프로토콜 방식
- 컨슈머가 새로 합류하면 리밸런싱이 트리거됨
- 컨슈머 그룹 내의 컨슈머들은 합류 요청과 자신들이 컨슘하는 토픽의 파티션 정보를 코디네이터에 전송함
- 코디네이터가 정보를 조합해 리더 컨슈머에 전송함
- 리더 컨슈머가 제외해야할 파티션 정보를 담아서 그룹 멤버들에게 전달함
- 그룹 멤버들은 필요없는 파티션을 골라 제외함. 만약 제외할 게 없으면 아무 작업을 하지 않음
- 제외된 파티션 할당을 하려고 컨슈머들이 다시 합류 요청을 보냄
- 리더 컨슈머는 제외된 파티션을 적절히 컨슈머들에 할당함
정확히 한 번(Exactly-once) 컨슈머 동작
- 컨슈머는
isolation.level옵션에read_committed값을 주면 트랜잭션이 완료된 메시지만 읽을 수 있게 됨 - 하지만, 컨슈머의 이 옵션은 트랜잭션 프로듀서가 보낸 메시지만 가져올 수 있는지만 보장함
- 컨슈머는 트랜잭션 코디네이터와 통신을 하지 않기 때문에 정확히 한 번만 메시지를 가져오는지 보장할 수 없음
- 컨슈머에 의해 컨슘된 메시지가 다른 싱크 저장소로 중복 저장될 수 있음
- 컨슈머가 정확하게 한 번 메시지를 가져오더라도 다른 애플리케이션에 저장하는 과정에서 중복 처리 될 수 있음
- 컨슈머의 동작까지 정확히 한 번 처리가 가능하려면 ‘Consume-Processing-Producing’이 모두 하나의 트랜잭션으로 처리되어야함
sendOffsetsToTransaction()메소드를 사용하면 컨슈머 그룹의 오프셋까지 트랜잭션에 추가할 수 있음
- 컨슈머 애플리케이션 레벨에서 정확히 한 번을 지원하는 경우가 있으니 확인바람!
LOG-INFO
끄적끄적
공식문서를 살펴보니 3버전부터는 기본적으로 컨슈머 파티션 전략은 레인지와 협력적 스티키 두 개가 디폴트 설정으로 잡혀있다고 합니다. 디폴트는 레인지고, 레인지를 리스트에서 제거하면 협력적 스티키가 된다고 하네요. 공식 도큐먼트
exactly once는 트랜잭션 코디네이터와 통신하는 부분은 따로 없고, 트랜잭션이 완료된 메시지만 읽을 수 있다. 기본적으로 read_uncommitted이고 추후 read_committed로 옵션을 변경하면 트랜잭션이 완료된 메시지만 읽을 수 있다.
컨슈머 자체에는 트랜잭션을 보장하는 코디네이터와의 통신이 없으므로 컨슈머 자체만의 트랜잭션 처리로는 중복이 발생할 여지가 존재한다.
따라서
sendOffsetsToTransaction메소드를 이용해, 컨슈머 그룹의 오프셋 커밋을 트랜잭션에 포함시켜서, 컨슈머-메시지 처리-프로듀싱 동작이 모두 하나의 트랜잭션으로 동작하게 되야한다.