[카프카 핵심 가이드] Ch 04

Ch 04. 카프카 컨슈머: 카프카에서 데이터 읽기

 

p.88

컨슈머는 해당 컨슈머 그룹의 그룹 코디네이터 역할을 지정받은 카프카 브로커에 하트비트를 전송함으로써 멤버십과 할당된 파티션에 대한 소유권을 유지한다.

 

Q. 그룹 코디네이터와 컨트롤러의 차이는?

 

 

정리하자면 카프카의 단일 브로커는 컨트롤러, 그룹 코디네이터 (복수 개의 group 담당 가능), 파티션의 리더 혹은 팔로워의 세가지 역할을 동시에 수행할 수 있다.

단, Kafka 3.x부터는 KRaft에서 컨트롤러 전용 노드 분리가 가능하다.

 

 

p. 89

 

Q. 컨슈머가 죽었다고 판단해 새로운 컨슈머에 파티션을 할당했으나, 기존 컨슈머가 살아있었고 동일한 레코드를 중복 처리하게 되는 케이스가 발생 가능한가?

 

A. 아래 시나리오와 같이 가능하다.

 

1. Consumer-1은 정상적으로 메시지를 처리 중 그러나 GC, 네트워크 지연, I/O 블로킹 등으로 heartbeat를 일정 시간 이상 보내지 못함

2. Group Coordinator는 session.timeout.ms를 넘겼다고 판단 → Consumer-1을 죽었다고 판단 리밸런싱 발생 → 해당 파티션은 Consumer-2에게 재할당

3. Consumer-1은 그 후에 깨어나서 계속 같은 파티션의 메시지를 처리함

4. 결과: 같은 오프셋 범위가 Consumer-1과 Consumer-2 양쪽에서 처리됨 → 중복 처리 발생

 

 

Q. 위 시나리오에서 컨슈머 1은 어떻게 살아있다고 다시 인식되는가?

 

A. Consumer-1은 Group Coordinator에게 "새로운 멤버로 재참여"한다.

 

예시

1. Consumer-1이 GC or 네트워크 이슈로 session.timeout.ms 이상 동안 heartbeat 미전송

2. Consumer-1이 회복됨

3. heartbeat 재시도 혹은 poll 재시작

4. 하지만 세션은 이미 만료되어 Group Coordinator는 Consumer-1을 새로운 멤버로 인식하여 JOIN_GROUP 요청

 

 

Q. 그럼 commitSync는 이미 퇴출된 컨슈머에 의해서도 반영될 수 있는가?

 

A. 컨슈머가 "이미 죽은 걸로 간주"되어 리밸런싱이 발생한 후에도, Coordinator는 뒤늦게 도착한 commit 요청을 반영할 수 있다.

consumer가 아직 그룹에 있어야만 offset commit을 할 수 있다는 강제 조건이 없고, late commit을 유연하게 수용하기 위한 설계이다.

 

 

 

p. 89

 

Q. 정적 그룹 멤버십과 timeout, heartbeat 등의 설정을 길게 유지하는 것의 차이는?

 

 

정적 멤버십의 장점

1. 재시작, 일시적 장애 시 리밸런싱 없이 복귀 가능

• 예 : 컨테이너 재기동, rolling deploy

2. 기존 파티션 -> 동일 Consumer에 할당

• 상태 기반 처리(stateful consumer)에서 매우 중요
• Kafka Streams, RocksDB 등을 쓰는 경우 매우 효과적

3. 정적 멤버십 + 충분한 session.timeout.ms 조합

• 일시적 장애에서 리밸런싱 없이 회복 가능

 

 

Q. 정적 멤버십을 사용하는 컨슈머가 timeout 끝나고 복귀하면?

 

A. 새 멤버로 간주되어 리밸런싱 발생하며, 다른 컨슈머가 동일한 id로 join한 상태라면 MemberIdRequiredException을 일으키거나 기존 멤버를 강제로 퇴출한다.

 

 

 

p. 94

그렇기 때문에 현재 컨슈머 코드에서 레코드를 읽어오지 않고 메타데이터만 가져오기 위해 poll(0)을 호출하고 있다면(상당히 일반적으로 쓰이는 우회 방법이다), 이를 poll(Duration.ofMillis(0))로 바꾼다고 해서 같은 결과를 기대할 수 없는 것이다.

 

Q. 메타데이터만 가져오는 poll 이란?

 

A. consumer.poll() 호출 시, 실제 메시지는 없지만, Kafka 브로커로부터 partition, offset, group 상태 등의 메타데이터 정보만 받아오는 상황이다.

 

대표적인 메타데이터 poll 케이스는 아래와 같다.

 

 

 

p. 95

fetch.max.bytes
브로커가 컨슈머에 레코드를 보낼 때는 배치 단위로 보내며, 만약 브로커가 보내야 하는 첫 번째 레코드 배치의 크기가 이 설정값을 넘길 경우, 제한 값을 무시하고 해당 배치를 그대로 전송한다.

 

Q. 최대 설정값이 넘어도 전송 가능한가?

 

A. 맞다. Kafka에서 fetch.max.bytes는 엄격한 hard limit이 아니라 소프트 상한선이라 브로커는 설정된 최대값을 초과하더라도 배치 단위로는 예외적으로 초과 전송할 수 있다.

 

 

p. 102

앞에서 설명한 것과 같이, 카프카의 고유한 특성 중 하나는 많은 JMS 큐들이 하는 것처럼 컨슈머로부터의 응답을 받는 방식이 아니라는 점이다.

 

Kafka

• 메시지를 보내고 나면 브로커는 "이 메시지를 누가 읽었는지" 알지 못한다
• Consumer가 읽었는지, 실패했는지, 중복 처리했는지는 브로커가 관여하지 않고, 오로지 Consumer가 offset을 어디까지 commit 했는지만을 기준으로 판단한다

JMS Queue

• Consumer가 처리 후 ack를 보내야만 메시지가 삭제된다
• ack가 없으면 다시 전송하거나 DLQ로 이동한다

 

 

p. 104

 

Q. consumer.commitSync()와 acknowledgement.acknowledge()의 차이는?

 

A. 둘 다 offset 커밋이라는 관점에서는 같다

acknowledgement.acknowledge()는 Spring Kafka Listener에서 추상화된 개념이고, 내부적으로 consumer.commitSync() 또는 commitAsync()를 호출한다.

consumer.commitSync()는 Kafka Consumer API 이며 로우레벨이다.

 

 

p. 112

 

Q. seek(), poll() 의 차이는?

 

A. seek()은 어떤 오프셋부터 메시지를 읽을지 위치를 지정하는 메서드이고, poll()은 지정된 위치부터 메시지를 실제로 가져오는 메서드이다.

 

 

Q. 그렇다면 seek() 호출 시 __consumer_offsets 토픽에 무언가 기록되거나 전송되는가?

 

A. 아니다.

seek()은 Consumer 인스턴스의 읽기 커서를 바꾸는 클라이언트 측 연산이다.

Kafka 브로커나 __consumer_offsets 토픽에는 아무것도 전송하지 않는다.

 

 

 

p. 121

컨슈머가 그룹에 조인할 필요가 없으니 subscribe() 메서드를 호출할 일이야 없겠지만, 오프셋을 커밋하려면 여전히 group.id 값을 설정해줄 필요가 있을 것이다.

 

Kafka에서 offset 커밋은 브로커의 내부 토픽인 __consumer_offsets에 다음과 같은 key로 저장됩니다:

(group.id, topic, partition) → offset

 

즉, Kafka 브로커 입장에서 "이 커밋은 누구의 오프셋인지"를 식별할 수 있어야 저장할 수 있습니다.
→ 따라서 실제로 그룹 참여(subscribe → rebalance)는 하지 않아도, 커밋 자체에는 group.id가 필수입니다.