Kafka의 Consumer Group과 Rebalance: Consumer가 멈추는 이유

1. 개요

당신의 서비스가 Kafka에서 메시지를 읽으며 여러 날 동안 안정적으로 실행되고 있습니다. 그런데 어느 날 consumer lag이 급증합니다. 로그에는 Revoking previously assigned partitions과 CommitFailedException으로 가득합니다. 주범은 보통 consumer group의 rebalance입니다.

이 글은 consumer group이 어떻게 partition을 분배하는지, rebalance가 언제 발생하는지, 왜 consumer를 “멈추게” 하는지 설명합니다. 읽고 나면 Spring Kafka 설정으로 rebalance 횟수를 줄이고 각 멈춤 시간을 단축하는 방법을 알게 될 것입니다.

2. Consumer Group은 어떻게 작동하는가

consumer group은 같은 group.id를 사용하는 consumer들의 모음입니다. Kafka는 topic의 partition들을 group의 구성원들에게 분배합니다. 각 partition은 한 번에 하나의 consumer에게만 할당됩니다.

orders topic이 6개의 partition을 가지고 있다고 가정하면, order-service의 3개 인스턴스를 실행할 때:

order-service-1  -> orders-0, orders-1
order-service-2  -> orders-2, orders-3
order-service-3  -> orders-4, orders-5

7개의 인스턴스를 실행하면 할당할 partition이 없어져서 한 인스턴스는 유휴 상태가 됩니다.

먼저 의존성을 추가하세요:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Spring Boot 3.x에서의 간단한 listener 예제:

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("주문 {} 수신 (partition {})", record.key(), record.partition());
    }
}

@KafkaListener annotation은 listener container를 생성합니다. 컨테이너 내부에는 poll() 루프를 실행하는 thread가 있습니다. groupId = "order-service"를 사용하는 모든 인스턴스는 같은 group에 참여합니다. 한 broker는 group coordinator 역할을 하여 구성원 목록과 partition 분배를 관리합니다.

3. Rebalance란 무엇이고 언제 발생하는가

Rebalance는 group coordinator가 partition을 구성원들에게 다시 분배하는 프로세스입니다. Rebalance를 유발하는 상황:

1. 새 consumer가 참여: 스케일 업, 새 인스턴스 배포.
2. Consumer가 그룹에서 주도적으로 떠남: 종료, rolling restart.
3. Consumer가 session.timeout.ms 내에 heartbeat를 보내지 않아 죽은 것으로 간주됨. 기본값은 45초입니다.
4. Consumer가 max.poll.interval.ms 내에 poll()을 호출하지 않음. 기본값은 5분입니다.
5. Topic의 partition 수가 변경되거나, 또는 subscription이 pattern과 일치하는 새 topic을 발견.

항목 3과 4의 두 timeout 메커니즘은 서로 다르며 자주 혼동됩니다:

설정	기본값	확인 내용
heartbeat.interval.ms	3000	Heartbeat 송신 주기 (백그라운드 thread에서)
session.timeout.ms	45000	프로세스가 살아있는지
max.poll.interval.ms	300000	처리 thread가 정체하지 않았는지

Heartbeat는 별도의 백그라운드 thread에서 전송됩니다. 따라서 처리 thread가 정체되어 있어도 프로세스는 “살아있습니다”. 반대로 max.poll.interval.ms는 메시지 처리 코드가 너무 느린 경우를 정확히 감지합니다.

4. Consumer가 멈추는 이유

4.1. Eager Rebalance: 전체 Group 정지

전통적인 프로토콜과 eager 타입의 assignor (RangeAssignor, RoundRobinAssignor, StickyAssignor)를 사용하면, 각 rebalance는 다음과 같이 진행됩니다:

1. 모든 consumer는 보유한 partition을 모두 반환합니다.
2. 모든 consumer는 JoinGroup 요청을 다시 보내고 서로를 기다립니다.
3. Group의 leader가 새로운 분배 계획을 계산하고, coordinator가 결과를 전송합니다.
4. Consumer는 새 partition을 받고 처리를 계속합니다.

1-3 단계가 진행되는 동안 group의 누구도 메시지를 처리하지 않습니다. 새로 시작하는 인스턴스 하나만으로도 전체 group이 정지됩니다. 이를 “stop-the-world” 현상이라고 합니다.

4.2. 처리가 너무 느림: max.poll.interval.ms 초과

실제로 가장 흔한 원인은 코드 처리가 느린 경우입니다. 다음 예는 각 메시지마다 외부 서비스 호출에 약 1초가 걸립니다:

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // 약 1초 소요
}

기본값은 max.poll.records = 500입니다. 하나의 poll() 호출은 최대 500개의 record를 반환할 수 있으며, 이는 약 500초의 처리 시간을 의미합니다. 이 수치는 max.poll.interval.ms의 300초를 초과합니다. 이 경우 consumer는 자동으로 group을 떠나고 다음과 같은 로그를 볼 수 있습니다:

[Consumer clientId=consumer-order-service-1, groupId=order-service] consumer poll timeout has expired.
This means the time between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages. You can address this either by increasing max.poll.interval.ms or by
reducing the maximum size of batches returned in poll() with max.poll.records.

그 후 offset commit이 실패합니다:

org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed
since the consumer is not part of an active group for auto partition assignment; it is likely
that the consumer was kicked out of the group.

뒤따르는 일련의 문제들:

• 이 consumer의 partition이 다른 곳에 재할당됩니다.
• Commit되지 않은 offset이 있으므로, 새 할당자는 해당 메시지들을 다시 처리합니다.
• 이전 consumer가 다시 참여하여 또 다른 rebalance를 유발합니다.
• 다음 batch도 여전히 느리면 이 루프가 계속 반복됩니다. Lag이 계속 증가합니다.

5. 해결 방법

5.1. Batch 크기와 poll 시간 조정

기억할 규칙: max.poll.records × 각 record의 처리 시간이 max.poll.interval.ms보다 훨씬 작아야 합니다.

spring:
  kafka:
    consumer:
      group-id: order-service
      max-poll-records: 50
      properties:
        max.poll.interval.ms: 120000

50개의 record와 record당 1초로, 하나의 batch는 약 50초가 소요됩니다. 이 수치는 120초 제한 내에서 안전합니다.

참고: 긴급 대응으로 max.poll.interval.ms를 몇 시간으로 늘리지 마세요. 이 값은 또한 Kafka가 정말 멈춘 consumer를 감지하는 최대 시간입니다. 또한 rebalance는 가장 느린 consumer의 poll() 호출을 기다려야 하므로, 큰 값은 rebalance를 더 길게 만듭니다.

메시지 하나의 처리가 원래 오래 걸리는 경우, 예를 들어 느린 API 호출이라면 외부 호출에 timeout을 설정해야 합니다. 다른 방법은 메시지를 내부 큐로 보내는 것이므로 poll thread를 너무 오래 점유하지 않습니다.

5.2. Cooperative Rebalancing

CooperativeStickyAssignor는 incremental rebalance를 사용합니다. Consumer는 정말 이동해야 하는 partition만 반환하고, 나머지 partition은 정상적으로 처리를 계속합니다.

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

order-service-4를 2절의 3개 인스턴스 group에 추가할 때, 일부 partition만 새 인스턴스로 이동합니다:

order-service-1  -> orders-0, orders-1        (그대로 유지, 정지 없음)
order-service-2  -> orders-2                  (orders-3 상실)
order-service-3  -> orders-4                  (orders-5 상실)
order-service-4  -> orders-3, orders-5

참고: 실행 중인 group에서 eager에서 cooperative로 직접 변경할 수 없습니다. Rolling deploy를 두 번 수행해야 합니다. 첫 번째는 둘 다 구성합니다: CooperativeStickyAssignor,RangeAssignor. 두 번째에는 RangeAssignor를 제거합니다. 구성원이 공통 assignor를 가지지 않으면 group은 오류를 보고합니다.

5.3. Static Membership

Rolling restart마다 인스턴스가 group을 떠났다가 다시 돌아옵니다. 결과적으로 각 pod마다 rebalance가 두 번 발생합니다. Static membership이 이 문제를 해결합니다. group.instance.id를 통해 각 인스턴스에 고정 ID를 부여합니다:

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: ${HOSTNAME}
        session.timeout.ms: 60000

group.instance.id를 가진 consumer는 종료 시 LeaveGroup을 보내지 않습니다. session.timeout.ms 내에 같은 ID로 돌아오면, coordinator는 rebalance 없이 이전 partition을 정확히 반환합니다.

참고: HOSTNAME은 재시작을 통해 안정적이어야 합니다(예: Kubernetes StatefulSet의 pod 이름). 일반 Deployment에서는 pod 이름이 매번 변경되므로 static membership은 효과가 없습니다. 같은 ID로 동시에 실행되는 두 인스턴스는 서로 group에서 제거합니다(FencedInstanceIdException). concurrency > 1일 때 Spring Kafka는 각 consumer에 -0, -1 등을 자동으로 추가합니다.

이 접근법의 트레이드오프: 인스턴스가 정말 죽으면, 그 partition은 session.timeout.ms가 끝날 때까지 처리되지 않습니다.

5.4. 새로운 Consumer Group 프로토콜

Kafka 4.0부터 새로운 rebalance 프로토콜(KIP-848)이 프로덕션 준비가 되었습니다. group.protocol=consumer로 활성화합니다. Partition 분배가 완전히 broker로 이동하고 항상 incremental 방식으로 진행됩니다. Consumer는 더 이상 공유 JoinGroup “배리어”를 통해 서로를 기다릴 필요가 없습니다.

이 프로토콜에서는 client 쪽의 partition.assignment.strategy가 더 이상 사용되지 않습니다. 대신 broker가 assignor를 선택합니다(client 쪽 구성은 group.remote.assignor). 당신의 cluster와 client가 충분히 새로우면 이것은 고려할 가치가 있는 선택입니다. Spring Boot가 가져오는 kafka-clients 버전을 사전에 확인해야 활성화할 수 있습니다.

6. Partition 반환 시 처리

Rebalance를 어떻게 최적화하든 rebalance는 항상 발생합니다. 중요한 것은 partition을 잃기 전에 처리된 offset을 commit하여 중복 처리를 줄이는 것입니다.

Spring Kafka는 partition이 반환되기 전에 대기 중인 offset을 자동으로 commit합니다. 직접 KafkaConsumer를 사용하면 ConsumerRebalanceListener를 사용하여 직접 이를 수행해야 합니다:

public class CommitOnRevokeListener implements ConsumerRebalanceListener {

    private final Consumer<String, String> consumer;
    private final Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();

    public CommitOnRevokeListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    public void markProcessed(ConsumerRecord<String, String> record) {
        pending.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
            OffsetAndMetadata offset = pending.remove(tp);
            if (offset != null) {
                toCommit.put(tp, offset);
            }
        }
        if (!toCommit.isEmpty()) {
            consumer.commitSync(toCommit);
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 여기서 새로운 partition에 대해 로컬 상태를 다시 로드할 수 있습니다
    }
}

각 부분 설명:

• markProcessed()는 record 처리 완료 후 다음에 읽을 offset, 즉 offset + 1을 기록합니다.
• onPartitionsRevoked()는 poll thread 자체에서 실행되며, partition이 제거되기 바로 전에 실행됩니다. 반환되는 partition의 offset만 commit합니다. Cooperative rebalancing에서는 일부 partition만 반환되므로 이것이 중요합니다.
• commitSync()는 broker 확인 때까지 차단하여 다른 consumer가 partition을 받기 전에 offset이 저장되도록 보장합니다.

Subscribe 시 listener를 등록합니다: consumer.subscribe(List.of("orders"), listener).

kafka-clients에 포함된 MockConsumer로 이 동작을 테스트합니다:

class CommitOnRevokeListenerTest {

    @Test
    void shouldCommitOnlyRevokedPartitions() {
        MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition p0 = new TopicPartition("orders", 0);
        TopicPartition p1 = new TopicPartition("orders", 1);
        consumer.assign(List.of(p0, p1));

        CommitOnRevokeListener listener = new CommitOnRevokeListener(consumer);
        listener.markProcessed(new ConsumerRecord<>("orders", 0, 41L, "order-1", "{}"));
        listener.markProcessed(new ConsumerRecord<>("orders", 1, 7L, "order-2", "{}"));

        listener.onPartitionsRevoked(List.of(p0));

        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(p0, p1));
        assertThat(committed.get(p0).offset()).isEqualTo(42L);
        assertThat(committed.get(p1)).isNull();
    }
}

Test는 orders-0만 반환되는 상황을 시뮬레이션합니다. 예상 결과는 orders-0에 대해 offset 42가 commit되는 것입니다. orders-1은 여전히 보유하고 있으므로 아직 commit된 offset이 없습니다.

참고: onPartitionsRevoked() 내의 코드는 빠르게 실행되어야 합니다. 이 callback은 rebalance 중에 실행됩니다. 느리면 전체 group이 기다려야 합니다.

7. 흔한 실수

• 비동기 처리하지만 poll에 따라 commit. Record를 ExecutorService로 보낸 후, container가 poll() 직후 offset을 commit합니다. Rebalance가 발생하면, 처리 중인 메시지가 손실될 수 있습니다. 완전히 처리된 record의 offset만 commit하세요.
• session.timeout.ms를 max.poll.interval.ms와 혼동. Session timeout을 늘려도 처리가 느린 경우는 도움이 되지 않습니다.
• concurrency를 partition 수보다 크게 설정. 추가 consumer는 할 일이 없습니다. 그들은 단지 group을 더 크게 만들고 rebalance를 더 길게 만듭니다.
• GC 일시 중지가 길거나 container에서 CPU가 제한됨. Heartbeat thread가 따라잡지 못하고, consumer는 코드가 느리지 않아도 group에서 제거됩니다. JVM 메트릭과 CPU 제한을 확인하세요.
• 처리를 멱등성으로 설계하지 않음. Rebalance는 항상 일부 메시지의 중복 처리를 야기할 수 있습니다. 트랜잭션을 사용하거나 중복 제거 메커니즘(예: 이미 처리된 orderId 확인)이 없으면 “정확히 한 번” 처리를 가정하지 마세요.

8. 결론

Rebalance는 consumer group이 탄력적으로 확장하고 오류를 견딜 수 있도록 하는 필요한 메커니즘입니다. 대가는 consumer가 일시적으로 정지해야 한다는 것입니다. Consumer를 지속적으로 “멈추게” 하는 가장 흔한 원인은 poll 루프가 max.poll.interval.ms를 초과하는 경우입니다. 이 경우 consumer는 group에서 제거되고, commit이 실패하며 rebalance가 계속 반복됩니다.

최소화하기 위해 다음을 수행할 수 있습니다:

• max.poll.records로 각 batch를 작게 유지.
• CooperativeStickyAssignor 또는 새로운 consumer group 프로토콜로 전환.
• 안정적인 식별자를 가진 배포에서 static membership 사용.
• onPartitionsRevoked()에서 올바르게 offset 커밋.

마지막으로 항상 메시지 처리를 멱등성으로 설계하세요. Rebalance는 완전히 제거할 수 없으며, 덜 자주 발생하게 하고 손상을 최소화하는 것만 가능합니다.