Kafka 컨슈머 그룹과 파티션 전략 – 처리량을 2배 높이는 설계법

[order-events 토픽]
  Partition 0 ──┬──→ billing-group    : Consumer A
                └──→ notification-group: Consumer X

  Partition 1 ──┬──→ billing-group    : Consumer B
                └──→ notification-group: Consumer Y

  Partition 2 ──┬──→ billing-group    : Consumer C
                └──→ notification-group: Consumer Z

__consumer_offsets (Kafka 내부 토픽)
┌─────────────────┬───────────┬─────────┬────────┐
│ Group ID        │ Topic     │ Partition│ Offset │
├─────────────────┼───────────┼─────────┼────────┤
│ billing-group   │ order-ev. │    0    │  1024  │
│ billing-group   │ order-ev. │    1    │   987  │
│ notify-group    │ order-ev. │    0    │   512  │
└─────────────────┴───────────┴─────────┴────────┘

파티션 3개, 컨슈머 3개 → 이상적인 1:1 배분 (최대 병렬 처리)

  P0 ──→ Consumer 1  ✅ 처리 중
  P1 ──→ Consumer 2  ✅ 처리 중
  P2 ──→ Consumer 3  ✅ 처리 중

파티션 3개, 컨슈머 5개 → 2개 컨슈머 유휴 (자원 낭비)

  P0 ──→ Consumer 1  ✅ 처리 중
  P1 ──→ Consumer 2  ✅ 처리 중
  P2 ──→ Consumer 3  ✅ 처리 중
         Consumer 4  💤 유휴 (파티션 없음)
         Consumer 5  💤 유휴 (파티션 없음)

파티션 3개, 컨슈머 1개 → 순차 처리 (병렬 처리 불가)

  P0 ┐
  P1 ├──→ Consumer 1  ⚠️ 혼자 3개 파티션 처리
  P2 ┘

필요 파티션 수 = max(목표 처리량 / 단일 파티션 처리량, 목표 처리량 / 단일 컨슈머 처리량)

예시:
  - 목표 처리량: 100,000 msg/s
  - 단일 파티션 쓰기 처리량(Producer 기준): 50,000 msg/s
  - 단일 컨슈머 처리량: 25,000 msg/s

  Producer 기준: 100,000 / 50,000 = 2개
  Consumer 기준: 100,000 / 25,000 = 4개
  → max(2, 4) = 파티션 4개 필요

브로커 3대, 파티션 6개:
  Broker 1: Partition 0, 3
  Broker 2: Partition 1, 4
  Broker 3: Partition 2, 5
  → 각 브로커가 2개씩 균등 분산 ✅

브로커 3대, 파티션 5개:
  Broker 1: Partition 0, 3
  Broker 2: Partition 1, 4
  Broker 3: Partition 2         ← 한 브로커만 1개 담당
  → 불균형 발생 ⚠️

// KafkaPartitionManager.java
@Component
public class KafkaPartitionManager {

    private final AdminClient adminClient;

    public KafkaPartitionManager(AdminClient adminClient) {
        this.adminClient = adminClient;
    }

    /**
     * 현재 토픽의 파티션 수 조회
     */
    public int getPartitionCount(String topicName) throws Exception {
        DescribeTopicsResult result = adminClient.describeTopics(
                Collections.singletonList(topicName));
        TopicDescription desc = result.topicNameValues()
                .get(topicName).get();
        return desc.partitions().size();
    }

    /**
     * 파티션 수 증가 (기존 파티션 수 → 목표 파티션 수)
     * ⚠️ 파티션 추가 후 기존 메시지의 파티션 배분은 변경되지 않음
     */
    public void increasePartitions(String topicName, int newPartitionCount) {
        Map<String, NewPartitions> newPartitions = Collections.singletonMap(
                topicName,
                NewPartitions.increaseTo(newPartitionCount)
        );
        adminClient.createPartitions(newPartitions);
        System.out.printf("토픽 '%s' 파티션 수를 %d개로 증가%n",
                topicName, newPartitionCount);
    }
}

리밸런싱 발생 트리거:
  ① 컨슈머 인스턴스 추가 (스케일 아웃)
  ② 컨슈머 인스턴스 제거 (스케일 인, 배포)
  ③ 컨슈머 장애 (session.timeout.ms 초과)
  ④ 컨슈머 처리 지연 (max.poll.interval.ms 초과)
  ⑤ 파티션 수 변경

리밸런싱 중 타임라인:
  t=0  : 컨슈머 C 장애 감지
  t=3s : 브로커가 그룹 리더(컨슈머 A)에게 리밸런싱 지시
  t=3s : 그룹 내 모든 컨슈머 처리 중단 ← Stop-The-World 시작
  t=8s : 새 파티션 할당 완료
  t=8s : 모든 컨슈머 처리 재개 ← 약 5초간 공백 발생

# application.yml
spring:
  kafka:
    consumer:
      properties:
        # 협력적 리밸런싱 활성화 (Kafka 2.4+ 권장)
        partition.assignment.strategy: >
          org.apache.kafka.clients.consumer.CooperativeStickyAssignor

# application.yml
spring:
  kafka:
    consumer:
      properties:
        # 각 컨슈머 인스턴스에 고유 ID 부여 (Pod 이름 등 활용)
        group.instance.id: order-consumer-pod-1
        # 세션 타임아웃을 길게 설정해 짧은 재시작에서 리밸런싱 방지
        session.timeout.ms: 60000

spring:
  kafka:
    consumer:
      properties:
        # 기본값 300,000ms(5분). 무거운 처리 작업이 있다면 늘릴 것
        max.poll.interval.ms: 600000   # 10분
        # 한 번 poll 시 가져오는 최대 레코드 수 줄이기 → 처리 시간 단축
        max.poll.records: 50

// GracefulShutdownConsumer.java
@Component
public class GracefulShutdownConsumer {

    private volatile boolean running = true;

    @KafkaListener(topics = "order-events", groupId = "order-group")
    public void consume(ConsumerRecord<String, OrderEvent> record,
                        Acknowledgment ack) {
        if (!running) return;
        try {
            processRecord(record.value());
            // 수동 커밋: 처리 완료 후 오프셋 커밋 (메시지 유실 방지)
            ack.acknowledge();
        } catch (Exception e) {
            // 처리 실패 시 커밋하지 않음 → 재시도 보장
            log.error("메시지 처리 실패: {}", record.key(), e);
        }
    }

    // 애플리케이션 종료 시 안전하게 컨슈머 중단
    @PreDestroy
    public void shutdown() {
        this.running = false;
        log.info("컨슈머 Graceful Shutdown 시작");
    }

    private void processRecord(OrderEvent event) {
        // 실제 비즈니스 로직
    }
}

[좋은 파티션 키 설계 예시]

케이스 1 – 주문 시스템: orderId를 키로 사용
  → 같은 주문의 이벤트(생성→결제→배송)가 항상 같은 파티션 → 순서 보장 ✅

케이스 2 – 사용자 활동 로그: userId를 키로 사용
  → 같은 사용자의 행동이 순서대로 처리 → 세션 분석 가능 ✅

케이스 3 – 센서 데이터: deviceId를 키로 사용
  → 같은 장치의 시계열 데이터가 순서 유지 → 이상 탐지 정확도 향상 ✅

[나쁜 파티션 키 설계 예시]

케이스 4 – 상태값(status)을 키로 사용: "CREATED", "PAID", "SHIPPED"
  → 값의 종류가 3개뿐 → 파티션이 10개라도 3개만 사용 → Hot Partition ⚠️

케이스 5 – 날짜(date)를 키로 사용: "2024-01-01"
  → 같은 날짜 트래픽이 특정 파티션에 집중 → 불균형 ⚠️

// HotPartitionResolver.java
// 해결책: 키에 랜덤 접미사를 붙여 파티션 분산 (순서 포기 시)
public String generateDistributedKey(String baseKey, int numPartitions) {
    // "userId_123" → "userId_123-2" (0~파티션수 사이 랜덤 접미사)
    int suffix = (int) (Math.random() * numPartitions);
    return baseKey + "-" + suffix;
}

// 해결책 2: 커스텀 파티셔너로 균등 분산
@Component
public class UniformPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int numPartitions = cluster.partitionCountForTopic(topic);
        if (keyBytes == null) {
            // 키가 없으면 라운드로빈 (균등 분산)
            return (int) (System.currentTimeMillis() % numPartitions);
        }
        // 키 해시값으로 파티션 결정 (기본 동작)
        return Math.abs(Utils.murmur2(keyBytes)) % numPartitions;
    }

    @Override public void close() {}
    @Override public void configure(Map<String, ?> configs) {}
}

// 동일 토픽을 3개의 컨슈머 그룹이 독립 소비
// ① 결제 서비스 컨슈머
@KafkaListener(topics = "order-events",
               groupId = "billing-group",
               concurrency = "3")  // 3개 스레드 → 파티션 3개와 매칭
public void billingConsumer(OrderEvent event) {
    billingService.processPayment(event);
}

// ② 알림 서비스 컨슈머
@KafkaListener(topics = "order-events",
               groupId = "notification-group",
               concurrency = "3")
public void notificationConsumer(OrderEvent event) {
    notificationService.sendAlert(event);
}

// ③ 분석 서비스 컨슈머 (배치 처리로 성능 최적화)
@KafkaListener(topics = "order-events",
               groupId = "analytics-group",
               concurrency = "3",
               containerFactory = "batchListenerFactory")
public void analyticsConsumer(List<OrderEvent> events) {
    analyticsService.processBatch(events);
}

Consumer Lag = Log End Offset - Current Offset

예시:
  Partition 0: Log End Offset = 10,000 / Current Offset = 9,500 → Lag = 500
  Partition 1: Log End Offset = 10,000 / Current Offset = 8,000 → Lag = 2,000 ⚠️
  Partition 2: Log End Offset = 10,000 / Current Offset = 9,900 → Lag = 100

  → Partition 1에 병목 발생, 해당 파티션 컨슈머 로직 점검 필요

# 컨슈머 그룹 전체 랙 조회
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group billing-group \
  --describe

# 출력 예시:
# GROUP           TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# billing-group   order-events     0           9500           10000        500
# billing-group   order-events     1           8000           10000       2000  ← 위험
# billing-group   order-events     2           9900           10000        100

# docker-compose.yml에 추가
  kafka-exporter:
    image: danielqsj/kafka-exporter:latest
    command:
      - "--kafka.server=kafka:9092"
    ports:
      - "9308:9308"
    depends_on:
      - kafka

컨슈머 랙이 지속 증가할 때 점검 순서:

Step 1. 파티션 수 확인
        → 컨슈머 수 > 파티션 수? → 파티션 증가 검토

Step 2. 컨슈머 로직 처리 시간 측정
        → 단일 메시지 처리에 수백 ms 이상? → DB 쿼리 최적화, 캐시 도입

Step 3. max.poll.records 조정
        → 기본 500개가 너무 많다면 줄여 처리 시간 단축
        → 배치 처리가 효율적이라면 늘려 처리량 향상

Step 4. concurrency 설정 확인
        → @KafkaListener concurrency = 파티션 수로 설정했는지 확인

Step 5. 외부 시스템 I/O 병목 확인
        → DB, 외부 API 호출이 병목? → 비동기 처리, 커넥션 풀 확장