Kafka Consumer Group ID를 제대로 설정하지 않으면 벌어지는 일

Kafka를 사용하는 서비스에서 이벤트가 발행된 뒤 Consumer가 바로 처리하지 않는 구간이 간헐적으로 보였습니다.

로그를 확인해 보니 메시지가 유실된 것은 아니었고, consumer lag이 계속 누적되는 상태도 아니었습니다. 처음에는 네트워크 문제나 특정 Consumer의 일시적인 처리 지연 정도를 의심했습니다.

그런데 Kafka 관련 로그를 확인하다 보니 리밸런싱이 생각보다 자주 발생하고 있었습니다.

Kafka의 Consumer Group에서는 하나의 파티션을 같은 그룹의 여러 Consumer가 동시에 나눠 읽지 않습니다. Group Coordinator가 현재 그룹에 어떤 Consumer들이 들어와 있는지 관리하고, 각 파티션을 어느 Consumer가 담당할지 할당합니다.

예를 들어 파티션이 4개이고 Consumer가 2개라면 대략 다음처럼 나눠 가집니다.

이 멤버 구성은 고정되어 있지 않습니다.

배포로 새로운 Consumer가 들어오거나 기존 Consumer가 종료될 수 있고, heartbeat나 max.poll.interval.ms 조건을 만족하지 못해 Kafka가 특정 Consumer를 그룹에서 제외할 수도 있습니다.

그룹의 멤버 구성이 바뀌면 기존 파티션 할당을 다시 계산해야 하는데, 이 과정이 리밸런싱입니다.

당시 그룹은 eager 방식으로 동작하고 있었습니다. 이 방식에서는 리밸런싱이 시작되면 Consumer들이 기존에 가지고 있던 파티션을 반납하고, 새로운 할당이 결정될 때까지 기다립니다.

이 사이에도 Producer가 발행한 이벤트는 broker의 파티션에 정상적으로 적재됩니다. 다만 새로운 Consumer가 할당되고 소비를 재개할 때까지 애플리케이션에서 실제로 이벤트를 처리하는 시점은 늦어질 수 있습니다.

처음 발견한 간헐적인 이벤트 처리 지연과 리밸런싱을 연결해서 보기 시작한 이유가 이것이었습니다.

그래서 Datadog에서 해당 Consumer Group의 상태를 조금 더 길게 확인했습니다.

최근 1시간만 봤을 때는 별다른 문제가 없어 보였지만, 일주일 기준으로 확인해 보니 generation이 16,453까지 올라가 있었습니다.

generation에는 절대적인 정상 수치가 있는 것은 아닙니다. 오래 운영된 그룹이거나 배포와 스케일링이 잦다면 자연스럽게 증가할 수 있습니다.

다만 이 그룹은 서비스 배포 빈도에 비해 리밸런싱이 지나치게 자주 발생하고 있었고, 로그에서도 그룹 멤버가 반복적으로 재구성되고 있었습니다.

그래서 실제 멤버를 확인해 보니 서로 독립적으로 개발하고 배포하는 여섯 개 서비스가 같은 Consumer Group ID를 사용하고 있었습니다.

여섯 서비스가 하나의 Consumer Group으로#

문제가 된 코드에서는 Consumer Group ID가 다음처럼 지정되어 있었습니다.

@KafkaListener(
    topics = "…",
    groupId = "consumerGroup"
)
public void consume(ConsumerRecord<?, ?> record) {
    // ...
}

각 서비스가 구독하는 토픽은 달랐지만 groupId 값은 모두 "consumerGroup"이었습니다.

처음에는 한 서비스의 설정 실수라고 생각했습니다. 그런데 같은 값을 사용하는 서비스를 몇 개 더 찾았고, Datadog에서 실제 그룹 멤버를 확인해 보니 서로 다른 여섯 서비스가 하나의 그룹에 들어가 있었습니다.

프로덕션 기준으로 약 90개의 Consumer가 하나의 그룹을 구성하고 있었습니다.

Kafka는 애플리케이션이나 Deployment의 경계를 알지 못합니다.

프로세스가 다르고 담당 팀이 달라도 group.id가 같으면 같은 Consumer Group의 멤버로 취급합니다.

즉 group.id는 단순한 이름이 아니라 파티션 할당과 그룹 멤버십을 어디까지 공유할지를 결정하는 설정입니다.

메시지는 처리됐지만 Consumer Group은 계속 리밸런싱됐다#

더 헷갈렸던 점은 이렇게 구성되어 있어도 대부분의 기능은 정상적으로 동작했다는 것입니다.

각 서비스가 서로 다른 토픽을 구독하고 있었기 때문입니다.

Kafka는 각 멤버의 subscription 정보를 알고 있으므로 알림 토픽을 구독하지 않는 CMS Consumer에 알림 파티션을 할당하지 않습니다.

따라서 이런 식의 오배치는 없었습니다.

알림 이벤트 → CMS 서비스
평가 이벤트 → 프로비저닝 서비스

각 서비스는 여전히 자신이 구독한 이벤트를 처리했습니다.

그래서 기능 테스트에서는 여러 서비스가 하나의 Consumer Group으로 묶여 있다는 사실이 잘 드러나지 않았습니다.

문제는 메시지가 어디로 전달되느냐가 아니라, 리밸런싱의 영향을 어디까지 함께 받느냐였습니다.

한 서비스의 배포가 여섯 서비스의 리밸런싱으로 이어졌다#

여섯 서비스는 서로 독립적으로 배포되고 있었습니다.

하지만 Kafka에서는 Consumer Group이 하나였습니다.

예를 들어 알림 서비스의 pod 하나가 재시작되면, Kafka 입장에서는 consumerGroup의 멤버 하나가 빠졌다가 다시 들어오는 것입니다.

eager 방식이면 원인 제공과 관계없는 멤버도 기존 파티션을 반납하고 새로운 할당을 기다릴 수 있습니다.

결과적으로 다른 서비스의 배포나 재시작 때문에 내 서비스의 Kafka 소비가 잠시 멈출 수 있는 구조였습니다.

애플리케이션의 배포 경계와 Kafka의 리밸런싱 경계가 맞지 않았던 것입니다.

리밸런싱이 많아지면 중복 처리 가능성도 커진다#

이벤트 처리가 잠깐 멈추는 것 외에 다른 영향도 확인해 봤습니다.

Kafka의 일반적인 at-least-once 소비에서는 애플리케이션의 이벤트 처리와 offset commit이 항상 하나의 원자적 작업으로 묶여 있지는 않습니다.

예를 들어 다음과 같은 상황이 가능합니다.

Consumer A가 이벤트를 처리했지만 아직 offset을 커밋하지 않은 상태에서 파티션이 회수되면, 새로 파티션을 받은 Consumer B는 마지막 committed offset부터 다시 읽을 수 있습니다.

이 경우 이미 처리한 메시지가 다시 전달될 수 있습니다. Kafka의 at-least-once 모델에서는 원래 고려해야 하는 상황입니다.

리밸런싱이 발생한다고 반드시 중복 처리가 생기는 것은 아니지만, 파티션 ownership이 자주 변경될수록 이런 상황이 발생할 수 있는 구간도 늘어납니다.

그래서 같은 시간대의 애플리케이션 로그도 확인했습니다.

리밸런싱이 많던 시간대에 중복 INSERT도 몰려 있었다#

프로덕션 로그를 하루 단위로 확인했을 때 18~19시 사이에 리밸런싱 관련 이벤트가 크게 증가한 구간이 있었습니다.

같은 시간대에 평가 서비스에서는 동일한 요약 점수 데이터를 다시 INSERT하려다 복합 유니크 제약에 걸린 오류가 12건 발생했습니다.

Duplicate entry '...'
for key '<복합 유니크 키>'

해당 구간에는 실제로 파티션 ownership이 한 pod에서 다른 pod로 이동하는 것도 확인할 수 있었습니다.

다만 이 12건 모두가 리밸런싱 때문에 발생했다고 1:1로 증명한 것은 아닙니다.

이를 명확하게 확인하려면 이벤트 ID와 partition, offset, Consumer instance, revoke/assign 시점을 하나의 trace로 연결해야 합니다.

이번에 확인한 것은 다음 정도였습니다.

리밸런싱이 집중된 시간대
        +
실제 partition ownership 이동
        +
같은 시간대 동일 데이터의 중복 INSERT

Kafka의 at-least-once 동작과도 일치하기 때문에 강한 원인 후보로 볼 수 있었지만, 상관관계 이상으로 단정하지는 않았습니다.

DB의 유니크 제약 덕분에 발견된 중복만 있다는 점도 신경 쓰였습니다.

알림 발송이나 외부 API 호출처럼 두 번 실행되어도 오류가 발생하지 않는 부수효과는 로그 없이 중복 실행될 수도 있기 때문입니다.

파티션을 받지 못한 Consumer도 있었다#

Consumer Group의 멤버를 확인하면서 다음과 같은 Consumer도 확인했습니다.

partitions=[]

할당받은 파티션이 하나도 없는 Consumer입니다.

각 서비스가 서로 다른 토픽을 구독하고 있고, 일부 토픽은 파티션 수보다 Consumer 인스턴스가 많아 처리할 파티션을 받지 못한 멤버가 존재했습니다.

이 Consumer들이 실제 메시지를 처리하는 것은 아니지만, 같은 그룹의 멤버인 이상 Group Coordinator와 heartbeat를 주고받고 그룹의 멤버십 변화에도 영향을 받습니다.

따라서 핵심 문제는 유휴 Consumer의 존재 자체보다 서로 독립적인 서비스들이 불필요하게 하나의 그룹 멤버십을 공유하고 있다는 점이었습니다.

공유한 group.id가 장애로 드러나지 않았던 이유#

가장 큰 이유는 기능 자체가 동작했기 때문이라고 봤습니다.

각 서비스는 자기 토픽을 정상적으로 소비했고, 메시지가 다른 서비스로 잘못 전달되지도 않았습니다.

리밸런싱 때문에 소비가 일시적으로 멈추더라도 결국 다시 처리됐고, consumer lag도 지속적으로 쌓이지 않았습니다.

비용에서도 크게 드러나지 않았습니다.

파티션을 받지 못한 Consumer가 있다고 해서 별도의 pod가 추가되는 구조가 아니라, 기존 애플리케이션 프로세스 내부에서 실행되고 있었기 때문입니다.

시스템이 완전히 실패하는 대신 조금씩 불안정하게 동작하고 있었던 셈입니다.

높은 generation이라고 원인이 모두 같지는 않았다#

조사 과정에서 generation이 이번 그룹보다 훨씬 높은 다른 Consumer Group도 발견했습니다.

하지만 원인은 조금 달랐는데 그 그룹은 여러 서비스가 group.id를 공유하지 않았습니다. 대신 메시지 처리 시간이 길어지면서 max.poll.interval.ms 안에 다음 poll()을 호출하지 못했고, Consumer가 그룹에서 제외됐다 다시 합류하는 과정이 반복되고 있었습니다.

그래서 generation은 원인을 직접 알려주는 값이라기보다 Consumer Group을 더 확인해 봐야 한다는 신호로 보는 편이 적절했습니다.

실제 리밸런싱 빈도와 멤버 구성, join/leave 원인, poll latency를 함께 봐야 원인을 구분할 수 있었습니다.

서비스마다 Consumer Group을 분리하다#

이번 문제의 해결 방향은 명확했습니다.

서로 독립적으로 개발하고 배포하는 서비스라면 Consumer Group도 서비스별로 분리했습니다.

변경 후에는 Consumer Group의 멤버 구성과 리밸런싱 로그를 다시 확인했습니다.

이전에는 한 서비스의 pod가 재시작되거나 배포될 때 다른 서비스까지 같은 그룹에서 리밸런싱에 참여했습니다.

분리 이후에는 각 서비스의 멤버 변화가 해당 Consumer Group 안에서만 처리됐고, 리밸런싱 발생 빈도와 generation 증가 양상도 이전보다 완화되는 것을 확인했습니다.

물론 Consumer Group을 분리한다고 리밸런싱 자체가 없어지는 것은 아닙니다.

각 서비스의 배포나 스케일링, Consumer 처리 지연으로 자기 그룹 안에서는 여전히 리밸런싱이 발생할 수 있습니다.

달라진 것은 다른 서비스까지 영향을 받던 구조가 사라졌다는 점입니다.

cooperative rebalance와 static membership은 영향만 줄인다#

리밸런싱의 영향을 줄이기 위해 cooperative rebalance나 static membership도 사용할 수 있습니다.

Cooperative 방식은 필요한 파티션을 점진적으로 이동시켜 eager 방식보다 소비 중단 범위를 줄일 수 있고, static membership의 group.instance.id는 Consumer 재시작 과정에서 불필요한 멤버 변화를 줄이는 데 도움이 됩니다.

하지만 이번 문제에서는 완화책에 가깝습니다.

두 방식을 적용해도 여섯 서비스가 같은 group.id를 사용하는 구조는 그대로입니다.

이번에는 리밸런싱 방식보다 먼저 Consumer Group의 경계를 서비스 경계에 맞추는 것이 필요했습니다.

Consumer Group은 함께 배포할 서비스의 경계였다#

이전에는 group.id를 Consumer를 구분하는 이름 정도로 생각했습니다. 이번 사례에서 확인한 group.id는 파티션 할당뿐 아니라 멤버 변화와 리밸런싱의 영향을 함께 받을 범위를 정하는 설정이었습니다. 같은 파티션을 나눠 처리할 필요가 없고 배포와 확장도 따로 하는 서비스라면 Consumer Group 역시 분리해야 합니다.

이후에는 generation 값만으로 리밸런싱 원인을 판단하지 않고, 같은 group.id에 어떤 서비스가 참여하는지와 실제로 같은 파티션을 나눠야 하는지를 먼저 확인하고 있습니다. 멤버 구성이 의도한 서비스 경계와 맞다면 join과 leave 원인, poll 간격과 처리 시간을 확인해 다음 원인을 좁힐 수 있습니다.

설정 단계에서는 서비스명을 포함한 group.id를 기본값으로 두어 의도하지 않은 공유를 줄일 수 있습니다.

spring:
  kafka:
    consumer:
      group-id: ${spring.application.name}-consumer

이후에는 topic과 partition 수만 정하는 데서 끝내지 않고, 같은 group.id를 사용할 서비스가 실제로 함께 배포되고 복구되어야 하는지까지 확인하고 있습니다.