[Kafka] Consumer 동작 방식과 리밸런싱

Consumer 동작 방식

1. Kafka Consumer가 Topic을 subscribe 합니다.
2. Kafka Consumer는 poll()을 반복 호출하면서 자신에게 할당된 partition의 메시지를 가져옵니다.
   ※ 반복 호출은 고정된 시간 간격으로 실행되는 것이 아닌, 이전 poll()로 가져온 record들을 처리한 뒤 다시 poll()을 호출하는 방식
3. partition에서 메시지를 가져온 뒤 commit을 통해 __conusmer_offset에 다음에 읽을 offset을 기록합니다.
    

[auto.offset.reset]
Consumer Group에 읽을 offset이 없거나 기존 offset이 유효하지 않을 때 어디서부터 읽기 시작할지를 정하는 설정입니다. 같은 Consumer Group 안에서 Consumer를 껐다 켜는 경우에는, 보통 저장된 offset이 있으므로 기존 위치부터 이어서 읽습니다.

즉, auto.offset.reset은 Consumer 재시작 옵션이 아니라, 새 Consumer Group이 처음 읽거나 기존 offset을 사용할 수 없을 때 적용되는 시작 위치 정책입니다.

- latest: 읽을 offset이 없으면 가장 최신 위치부터 읽습니다. 즉, 보통 이후에 들어오는 새 메시지부터 소비합니다. (default)
- earliest: 읽을 offset이 없으면 가장 오래된 메시지부터 읽습니다. 과거 데이터까지 모두 처리해야 할 때 사용합니다.
- none: 읽을 offset이 없으면 예외를 발생시킵니다. 자동으로 시작 위치를 정하지 않고 싶을 때 사용합니다.

Consumer Group

Consumer Group은 여러 Consumer를 하나의 논리적 작업 단위로 묶는 개념입니다. Kafka는 Consumer Group을 기준으로 메시지 소비를 관리하고, 파티션을 각 Consumer에 분배합니다. Consumer Group은 group.id로 식별됩니다.

 

동일한 group.id를 사용하는 Consumer들은 같은 Consumer Group으로 묶입니다. 같은 Consumer Group 내의 Consumer들은 파티션을 최대한 균등하게 분배받아 작업을 나눠 수행합니다. 서로 다른 group.id를 사용하는 Consumer들은 서로 완전히 독립적으로 동작합니다. 즉, 같은 Topic을 읽더라도 각 그룹은 자신의 offset을 별도로 관리합니다.

Consumer Group과 Consumer의 관계 (1:N)

하나의 Consumer Group은 여러 Consumer를 가질 수 있으며, 하나의 Consumer는 동시에 하나의 Consumer Group에만 속할 수 있습니다. Consumer는 group.id를 기준으로 어떤 Consumer Group에 속할지가 결정됩니다.
 
같은Consumer Group에 속한 Consumer들은 토픽의 파티션을 나누어 처리하고, 하나의 파티션은 같은 그룹 내에서 하나의 Consumer에게만 할당됩니다. 따라서 같은 파티션의 메시지가 여러 Consumer에 의해 중복 처리되지 않으며, Kafka는 이를 통해 작업 분산과 병렬 처리를 지원합니다.

Consumer Group은 언제 삭제될까?

Consumer Group은 모든 Consumer가 빠져 EMPTY 상태가 되어도 즉시 삭제되지 않습니다. Kafka는 그룹이 비어 있더라도 committed offset을 일정 기간 유지하며, 이 offset은 offsets.retention.minutes 설정(기본 값: 7일)에 따라 보관됩니다.

 

이후 offset이 만료되어 삭제되면, 해당 Consumer Group도 더 이상 유지할 정보가 없어 함께 정리될 수 있습니다. 즉, 순서상으로는 offset이 먼저 사라지고, 그다음 Consumer Group이 정리될 수 있습니다. 여기서 사라지는 것은 메시지 자체가 아니라, Consumer Group이 어디까지 읽었는지를 나타내는 진행 정보(offset) 입니다.

Consumer는 서버 한 대당 하나일까? 

서버 1대 = Consumer 1개는 가능한 구성 중 하나일 뿐이며, 실제로는 한 서버에서 여러 Consumer가 동시에 동작할 수도 있습니다. 예를 들어, Spring Kafka에서 @KafkaListener의 concurrency 값을 3으로 설정하면, 서버 한 대에서 3개의 KafkaConsumer 인스턴스가 생성되어 같은 Consumer Group에 참여합니다. 즉, Consumer는 서버 단위가 아니라 KafkaConsumer 인스턴스 단위로 동작한다고 이해하는 것이 더 정확합니다.
 
이 개념을 서버 수와 함께 보면 다음과 같습니다. 같은 group.id를 사용하는 Consumer 서버를 4대 띄우고, 각 서버마다 concurrency=3으로 설정했다면, 전체적으로는 3*4 = 12개의 Consumer 인스턴스가 하나의 Consumer Group에 참여하게 됩니다. 

(애플리케이션당 concurrency 3) * (서버 4대) = 12 컨슈머가 하나의 Consumer Group에 참여

 
다만 여기서 주의할 점은, Consumer 수가 많다고 해서 무조건 그 수만큼 병렬 처리되는 것은 아니라는 점입니다. 같은 Consumer Group에서는 하나의 파티션이 한 시점에 하나의 Consumer에게만 할당되므로, 실제 병렬 처리의 상한은 Consumer 수가 아니라 파티션 수에 의해 결정됩니다. 예를 들어 Consumer가 12개여도 파티션이 8개라면 실제로는 8개만 할당되고, 나머지 4개 Consumer는 대기 상태가 될 수 있습니다. 즉, 정리하면 Consumer Group의 병렬성은 단순히 서버 수나 concurrency 값만으로 결정되지 않고, 토픽의 파티션 수와 함께 고려해야 합니다.

Consumer Rebalancing이 발생하는 상황

Consumer Rebalancing은 Consumer Group 내 Consumer와 파티션 매핑을 다시 계산해 재배정하는 과정을 의미합니다.

 

Case1) Topic에 새로운 Partition이 추가될 때
Case2) Consumer Group 내에 새로운 Consumer가 추가되거나, 기존 Consumer가 종료될 때
Case3) session.timeout.ms 이내에 Heartbeat이 응답이 없을 때

• heartbeat.interval.ms 마다 Consumer가 Heartbeat를 보내고, 브로커는 마지막 하트비트 이후 session.timeout.ms 동안 하트비트가 안 오면 그 컨슈머를 죽었다고 판단해서 그룹에서 빼고 Rebalancing 합니다. (Heartbeat 전송 자체는 Consumer 클라이언트가 자동으로 처리)
• heartbeat.interval.ms는 일반적으로 session.timeout.ms의 1/3 이하로 두는 걸 권장합니다.
  • heartbeat.interval.ms 기본값: 3000ms (3초)
  • session.timeout.ms 기본값: 45000ms (45초)

Case4) max.poll.interval.ms 안에 poll( )이 호출되지 않을 경우 ⭐⭐⭐⭐⭐
max.poll.interval.ms는 이전 poll() 호출 이후, 다음 poll()이 다시 호출되기까지 허용되는 최대 시간입니다. 이 시간을 초과하면 Kafka는 해당 Consumer가 정상적으로 처리 루프를 돌지 못한다고 보고 Rebalancing이 발생합니다.
 
Consumer는 새 메시지가 없어도 poll()을 계속 호출하며, 메시지가 없으면 일정 시간 대기한 뒤 빈 결과를 반환합니다. 따라서 문제는 새로운 데이터 유무 자체가 아니라, 처리 로직이 오래 걸리거나 Consumer 스레드가 블로킹되어 다음 poll() 호출이 지연되는 상황에서max.poll.interval.ms(기본 값: 5분)를 초과해 Rebalance가 발생할 수 있습니다.

• max.poll.interval.ms는 commit 기준이 아니라 다음 poll() 호출 기준으로 초기화된다.

단건 listener를 사용한다고 해서 Kafka Consumer가 broker에서 record를 항상 1건씩만 가져오는 것은 아닙니다. Kafka Consumer는 poll() 한 번에 최대 max.poll.records 개수만큼 record를 반환할 수 있습니다. Spring Kafka의 record listener는 poll()로 가져온 record들을 listener 메서드에 한 건씩 전달하고, batch listener는 poll() 결과를 List 또는 ConsumerRecords 형태로 한 번에 전달합니다. 즉, record listener와 batch listener의 차이는 broker에서 record를 가져오는 단위가 아니라, poll()로 가져온 record를 listener 메서드에 전달하는 방식의 차이입니다.

 

이 때문에 단건 listener를 사용하더라도 한 번의 poll() 결과를 모두 처리하는 시간이 길어질 수 있습니다. 예를 들어 max.poll.records가 기본값인 500이고, record 한 건을 처리하는 데 시간이 오래 걸린다면 다음 poll() 호출이 지연될 수 있습니다. 이 지연 시간이 max.poll.interval.ms를 초과하면 Consumer가 정상적으로 polling하지 못한 것으로 판단되어 리밸런싱이 발생할 수 있습니다.

 

이를 방지하려면 한 번의 poll() 결과를 max.poll.interval.ms 안에 처리할 수 있도록 (1)max.poll.records 값을 줄여 한 번에 처리할 record 수를 제한할 수 있습니다. 반대로 record 처리 시간이 오래 걸릴 수밖에 없는 구조라면 (2)max.poll.interval.ms를 늘리는 것도 방법입니다.

max.poll.interval.ms
= poll() 호출 사이의 최대 허용 간격
= 너무 오래 처리해서 다음 poll()을 못 하면 리밸런싱 판단에 사용됨

pollTimeout
= poll() 한 번 호출했을 때 record를 기다리는 최대 대기 시간
= 메시지가 없을 때 poll()이 얼마나 기다릴지 결정

 

다만 근본적으로는 DB 처리, 외부 API 호출, 락 대기 등 record 한 건당 처리 시간을 줄이는 것이 중요합니다. AckMode.RECORD는 record 단위로 offset commit을 제어할 수 있어 실패 시 재처리 범위를 줄이는 데는 도움이 되지만, listener 처리 시간이 길어져 다음 poll() 호출이 늦어지는 문제 자체를 해결해주지는 않습니다.

*순수 KafkaConsumer를 직접 쓰는 경우
while 루프에서 poll(Duration.ofMillis(x))를 어떻게 호출하느냐에 따라 주기가 사실상 결정됩니다.
예를 들어 poll(100ms)면 더 자주 깨어나고, poll(5000ms)면 더 오래 기다릴 수 있습니다.

*Spring Kafka를 쓰는 경우
보통 listener container의 pollTimeout으로 한 번의 poll()이 얼마나 기다릴지를 조절합니다.
기본값은 5000ms(5초) 입니다

 
Rebalancing이 발생하면 그룹 코디네이터가 Consumer들의 구독 정보와 그룹 상태를 바탕으로 파티션을 다시 할당합니다. 이때 Classic(Eager) 방식은 기존 할당을 모두 해제한 뒤 다시 배정하고, Cooperative 방식은 필요한 파티션만 단계적으로 이동시켜 중단 시간을 줄입니다.