KO
FrançaisDeutschTiếng ViệtEspañolEnglishPortuguês日本語简体中文한국어
홈 / 아키텍처 & 설계 / Kafka

Kafka의 Consumer group과 rebalance (Spring Boot)

Cris The Dev · 2026. 10. 13. · 9 분 읽기 · 217

1. 개요

Kafka에서 같은 서비스의 여러 인스턴스를 실행할 때, 어느 인스턴스가 어떤 메시지를 처리할지 결정해야 합니다. Kafka는 consumer group과 rebalance 메커니즘으로 이 문제를 해결합니다.

이 글에서는 Kafka가 group 내의 partition을 어떻게 분배하는지, 그리고 rebalance가 발생하는 상황을 설명합니다. 그 다음 Spring Boot에서 rebalance 발생을 줄이고 데이터 손실이나 중복을 방지하도록 설정하는 방법을 알아봅니다. 글 끝에는 Embedded Kafka를 이용해 partition 분배를 검증하는 테스트가 있습니다.

2. Consumer group이 작동하는 방식

consumer group은 같은 group.id를 가진 consumer들의 모음입니다. Kafka는 다음을 보장합니다:

  • topic의 각 partition은 한 번에 group 내의 정확히 한 개 consumer에게만 할당됩니다.
  • 한 consumer는 여러 partition을 할당받을 수 있습니다.
  • 다른 group들은 독립적으로 읽습니다. 각 group은 자신의 offset을 가집니다.

topic orders가 4개의 partition을 가진다고 가정합시다:

Group "billing"에 2개의 consumer:
  consumer-A -> orders-0, orders-1
  consumer-B -> orders-2, orders-3

Group "billing"에 5개의 consumer:
  consumer-A -> orders-0
  consumer-B -> orders-1
  consumer-C -> orders-2
  consumer-D -> orders-3
  consumer-E -> (할 일 없음, partition 없음)

중요한 결과는 partition의 개수가 group의 최대 병렬성을 결정한다는 것입니다. partition 개수를 초과하는 consumer를 추가해도 처리 속도가 빨라지지 않습니다.

이 조정은 한 broker가 담당하며, 이를 group coordinator라고 합니다. Consumer는 살아있음을 알리기 위해 정기적으로 coordinator에 heartbeat를 보냅니다.

2.1. Dependency

Spring for Apache Kafka를 추가합니다. Spring Boot 3.x에서는 버전을 자동으로 관리합니다:

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

2.2. 간단한 listener

@Component
public class OrderListener {

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

    @KafkaListener(topics = "orders", groupId = "billing", concurrency = "2")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("partition={} offset={} order={}",
                record.partition(), record.offset(), record.value());
    }
}
  • groupId = "billing"은 consumer를 billing group에 배치합니다.
  • concurrency = "2"는 같은 인스턴스 내에서 2개의 consumer(2개의 스레드)를 생성합니다. 서비스의 2개 인스턴스를 실행하면, group은 4개의 consumer를 가지며, 각 consumer는 1개의 partition을 할당받습니다.

시작 시 로그는 partition 할당 결과를 보여줍니다:

billing: partitions assigned: [orders-0, orders-1]
billing: partitions assigned: [orders-2, orders-3]

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

Rebalance는 group이 partition을 멤버들 사이에 다시 분배하는 과정입니다. 다음과 같은 경우에 발생합니다:

  1. 새 consumer가 참여할 때, 예를 들어 pod을 scale out하거나 새 인스턴스를 배포할 때.
  2. Consumer가 정상적으로 나갈 때, 예를 들어 shutdown을 호출하거나 close()를 명시적으로 호출할 때.
  3. Consumer가 죽은 것으로 간주될 때: coordinator가 session.timeout.ms 내에 heartbeat를 받지 못할 때.
  4. Consumer가 처리하는 속도가 너무 느릴 때: 두 번의 poll() 호출 사이의 간격이 max.poll.interval.ms를 초과할 때. 이 경우 consumer가 자동으로 group을 나갑니다.
  5. Topic이 변경될 때: partition 개수를 늘리거나 구독 중인 패턴과 맞는 새 topic이 생길 때.

문제는 기존의 eager 프로토콜에서는 모든 consumer가 전체 partition을 반납한 후에야 다시 할당받습니다. 이 동안 전체 group의 처리가 중단되는데, 이를 흔히 “stop-the-world”라고 부릅니다. 큰 group이거나 rolling 배포에서는 rebalance가 계속 반복될 수 있습니다.

4. 중요한 timeout 파라미터들

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: billing
      enable-auto-commit: false
      max-poll-records: 100
      properties:
        session.timeout.ms: 45000
        heartbeat.interval.ms: 3000
        max.poll.interval.ms: 300000
파라미터 의미
session.timeout.ms heartbeat가 없는 최대 시간, 이후 consumer가 제거됩니다
heartbeat.interval.ms heartbeat 전송 주기, 보통 session.timeout.ms의 1/3 이하
max.poll.interval.ms 연속된 두 번의 poll() 호출 사이의 최대 간격
max-poll-records poll() 호출당 반환되는 최대 record 개수

Heartbeat는 별도의 백그라운드 스레드에서 전송됩니다. 따라서 메인 스레드가 바쁜 상태에서도 consumer는 session.timeout.ms 내에 “살아있다고” 인식됩니다. 하지만 max.poll.interval.ms는 처리 스레드만 계산합니다. 한 batch를 처리하는 데 시간이 너무 오래 걸리면 consumer가 group에서 제거됩니다.

로그에서 전형적인 오류는 다음과 같습니다:

Member consumer-billing-1 sending LeaveGroup request to coordinator due to
consumer poll timeout has expired. This means the time between subsequent
calls to poll() was longer than the configured max.poll.interval.ms ...
CommitFailedException: Commit cannot be completed since the group has already rebalanced

다음 공식을 만족해야 합니다:

max-poll-records × 1개 record 처리 시간  <  max.poll.interval.ms

주의: max.poll.interval.ms를 몇 시간으로 늘려서 ’해결’하지 마세요. 실제로 멈춘 consumer가 partition을 오래 유지하면서 처리하지 않게 됩니다. 대신 max-poll-records를 줄이거나, 무거운 작업을 consume 스레드 밖으로 옮기세요.

5. Rebalance의 영향 줄이기

5.1. Cooperative rebalancing

CooperativeStickyAssignor를 사용하면, 실제로 이동해야 하는 partition만 수거됩니다. 다른 consumer들은 rebalance가 진행되는 동안에도 자신의 partition을 계속 처리합니다.

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

예: group이 A를 가지고 있고 (orders-0..3), 그 후 B가 참여합니다.

Eager:       A가 4개 partition을 모두 반납 -> 재할당: A {0,1}, B {2,3}
Cooperative: A는 {2,3}만 반납, {0,1}은 계속 처리 -> B가 {2,3} 수신

주의: 실행 중인 group을 한 번의 배포로 eager에서 cooperative로 직접 전환할 수 없습니다. 두 번의 rolling 배포가 필요합니다. 첫 번째 배포에서는 두 assignor를 모두 선언하세요. 예: CooperativeStickyAssignor,RangeAssignor. 두 번째 배포에서 RangeAssignor를 제거합니다.

Kafka 4.0부터 새로운 rebalance 프로토콜(KIP-848)이 GA 상태입니다. group.protocol=consumer로 활성화하면, broker가 partition 분배를 수행하고, rebalance는 기본적으로 증분 방식이 되며, client 쪽의 partition.assignment.strategy와 session.timeout.ms 파라미터는 더 이상 사용되지 않습니다. 클러스터가 지원한다면, 이것이 가장 먼저 고려할 옵션입니다.

5.2. Static membership

pod이 재시작될 때마다 consumer는 새로운 member.id를 받게 되고, 이것이 두 번의 rebalance를 트리거합니다(나감 후 다시 참여). Static membership은 consumer에 고정된 식별자를 할당합니다:

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

Consumer가 session.timeout.ms 내에 돌아오면, coordinator는 rebalance 없이 정확히 이전 partition들을 다시 할당합니다. 이 방식은 HOSTNAME이 안정적인 Kubernetes StatefulSet(billing-0, billing-1…)과 매우 잘 맞습니다.

주의: group.instance.id는 group 내에서 유일해야 합니다. concurrency > 1일 때, Spring Kafka는 자동으로 각 sub-consumer에 -n 접미사를 추가합니다. 두 인스턴스가 같은 ID를 사용하면, 나중에 들어온 인스턴스가 이전 인스턴스를 “제거”합니다(FencedInstanceIdException). pod 이름이 무작위인 Deployment에서는 static membership을 사용하지 마세요. 아무 이득이 없을 것입니다.

6. Rebalance 중 offset 처리

Rebalance의 가장 큰 위험은 중복 처리입니다. Consumer A가 record를 처리했지만 아직 commit하지 않았을 때, partition이 B로 이동하면, B는 마지막으로 commit된 offset부터 다시 읽게 됩니다.

Spring Kafka는 기본적으로 각 batch 후에 commit하고(AckMode.BATCH), partition이 수거되기 전에 대기 중인 offset을 자동으로 commit합니다. Buffer를 flush하거나 로그를 기록하는 등 개입이 필요하면 ConsumerAwareRebalanceListener를 사용합니다:

@Component
public class OrderRebalanceListener implements ConsumerAwareRebalanceListener {

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

    @Override
    public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
                                                Collection<TopicPartition> partitions) {
        log.info("Partition {}을 잃을 예정입니다. 메모리의 데이터를 flush합니다", partitions);
    }

    @Override
    public void onPartitionsRevokedAfterCommit(Consumer<?, ?> consumer,
                                               Collection<TopicPartition> partitions) {
        log.info("Partition {}을 commit하고 반환했습니다", partitions);
    }

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer,
                                     Collection<TopicPartition> partitions) {
        log.info("Partition {}을 할당받았습니다", partitions);
    }

    @Override
    public void onPartitionsLost(Consumer<?, ?> consumer,
                                 Collection<TopicPartition> partitions) {
        log.warn("Partition {}을 commit 전에 손실했습니다", partitions);
    }
}
  • onPartitionsRevokedBeforeCommit은 container가 offset을 commit하기 전에 실행됩니다. 미완료된 작업을 마무리할 수 있는 곳입니다.
  • onPartitionsRevokedAfterCommit은 offset이 이미 commit된 후에 실행됩니다.
  • onPartitionsLost는 consumer가 group에서 제거될 때 호출됩니다. 예를 들어 max.poll.interval.ms를 초과했을 때입니다. 이 때 partition이 이미 다른 consumer에게 속할 수 있으므로 commit하지 마세요.

Container factory에 listener를 등록합니다:

@Configuration
public class KafkaConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> consumerFactory,
            OrderRebalanceListener rebalanceListener) {

        var factory = new ConcurrentKafkaListenerContainerFactory<Object, Object>();
        configurer.configure(factory, consumerFactory);
        factory.getContainerProperties().setConsumerRebalanceListener(rebalanceListener);
        return factory;
    }
}

configurer.configure(...)는 application.yml의 모든 설정을 유지합니다. 다음 줄은 rebalance listener만 추가합니다.

주의: 아무리 잘 설정해도 Kafka는 일반 consumer에 대해 at-least-once만 보장합니다. 처리를 멱등성으로 만드세요. 예를 들어, PaymentService를 호출하기 전에 orderId가 이미 결제되었는지 확인합니다. rebalance가 절대 중복을 야기하지 않을 것이라고 가정하지 마세요.

7. Partition 할당 테스트

다음 테스트는 @EmbeddedKafka를 사용하여 같은 group의 두 consumer가 4개 partition을 균등하게 나누는지 검증합니다:

@EmbeddedKafka(partitions = 4, topics = "orders")
class ConsumerGroupAssignmentTest {

    @Test
    void twoConsumersInSameGroupShareAllPartitions(EmbeddedKafkaBroker broker) {
        Map<String, Object> props = KafkaTestUtils.consumerProps("billing", "false", broker);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        try (var consumerA = new KafkaConsumer<String, String>(props);
             var consumerB = new KafkaConsumer<String, String>(props)) {

            consumerA.subscribe(List.of("orders"));
            consumerB.subscribe(List.of("orders"));

            long deadline = System.currentTimeMillis() + 30_000;
            while (System.currentTimeMillis() < deadline
                    && !(consumerA.assignment().size() == 2 && consumerB.assignment().size() == 2)) {
                consumerA.poll(Duration.ofMillis(200));
                consumerB.poll(Duration.ofMillis(200));
            }

            Set<TopicPartition> all = new HashSet<>(consumerA.assignment());
            all.addAll(consumerB.assignment());

            assertThat(consumerA.assignment()).hasSize(2);
            assertThat(consumerB.assignment()).hasSize(2)
                    .doesNotContainAnyElementsOf(consumerA.assignment());
            assertThat(all).hasSize(4);
        }
    }
}

설명:

  • @EmbeddedKafka는 메모리 내에서 broker를 시작하고, 4개의 partition을 가진 topic orders를 생성합니다. Broker는 test 메서드의 파라미터로 직접 주입됩니다.
  • 두 개의 KafkaConsumer는 같은 group.id = "billing"을 사용합니다.
  • 루프에서 두 consumer에 poll()을 번갈아 호출합니다. Consumer는 poll() 내에서만 rebalance에 참여하기 때문입니다. consumer A만 poll하면 rebalance가 완료되지 않습니다. 두 consumer는 같은 스레드에서 실행되지만 서로 다른 객체이므로 “한 consumer는 한 스레드에서만 사용된다”는 규칙을 위반하지 않습니다.
  • Assertion들은 각 consumer가 2개의 partition을 가지고, 서로 겹치지 않으며, 총 4개의 partition을 모두 커버하는지 검증합니다.

이것은 또한 다양한 partition.assignment.strategy를 실험할 수 있는 좋은 방법입니다. consumer 3개를 추가하고, 한 consumer를 닫은 후 assignment() 변화를 관찰해보세요.

8. 흔한 오류

  • Consumer 개수가 partition 개수보다 많음: 남은 consumer들은 아무것도 하지 않습니다. concurrency를 늘려도 시스템이 빨라지지 않습니다.
  • Listener에서 동기 처리가 너무 오래 걸림: max.poll.interval.ms를 초과하면 rebalance 루프가 발생합니다. Consumer가 제거되고, 돌아오고, 같은 batch를 다시 받고, 다시 제거되는 악순환입니다.
  • onPartitionsLost에서 수동 commit: CommitFailedException을 유발하거나, 새 consumer의 offset을 덮어씌워 더 심각한 문제를 초래합니다.
  • 두 개의 다른 서비스에 같은 group.id 사용: 두 서비스가 메시지를 ‘나눠먹게’ 되어 각각 모든 메시지를 받지 못합니다. 각 서비스는 자체 group을 가져야 합니다.
  • 큰 group에서 eager 프로토콜로 rolling 배포: 각 pod 재시작이 두 번의 stop-the-world를 유발합니다. Cooperative, static membership, 또는 Kafka 4.x의 새 프로토콜을 사용하세요.

9. 결론

Consumer group은 Kafka 읽기를 수평으로 확장할 수 있게 해주며, 상한은 partition 개수입니다. Rebalance는 그 유연성에 대한 대가입니다. 멤버가 들어오거나 나가거나, 죽거나, 처리가 너무 느릴 때 발생합니다.

Rebalance의 피해를 최소화하려면 다음을 해야 합니다:

  • max.poll.interval.ms와 max-poll-records를 합리적으로 설정하세요.
  • CooperativeStickyAssignor를 사용하거나, Kafka 4.x에서는 group.protocol=consumer를 사용하세요.
  • 인프라가 허용할 때 static membership을 활성화하세요.
  • ConsumerAwareRebalanceListener를 통해 partition 라이프사이클을 처리하세요.

마지막으로, 항상 처리를 멱등성으로 설계하세요. At-least-once는 기본 보장이며, 이를 받아들여야 합니다.

kafkaspring-kafkaconsumer-grouprebalancespring-boot
저자의 앱Reflex Arena11 brain-speed games, live 1v1 duels, monthly seasonsiOS · Android

같은 주제의 글