# 컨슈머 랙
컨슈머 랙(LAG)은 토픽의 오프셋(LOG_END_OFFSET)과 컨슈머 오프셋(CURRENT_OFFSET)간의 차이다.
프로듀서는 계속해서 새로운 데이터를 파티션에 저장하고 컨슈머는 자신이 처리할 수 있는 만큼 데이터를 가져간다.
컨슈머 랙은 컨슈머가 정상 동작하는지 여부를 확인할 수 있기 때문에 컨슈머 애플리케이션을 운영한다면 필수적으로 모니터링 해야하는 지표이다.
컨슈머 랙은 컨슈머 그룹과 토픽, 파티션별로 생성된다.
1개의 토픽에 3개의 파티션이 있고 1개의 컨슈머 그룹이 토픽을 구독하여 데이터를 가져가면 컨슈머 랙은 총 3개가 된다.
프로듀서가 보내는 데이터양이 컨슈머의 데이터 처리량보다 크다면 컨슈머 랙은 늘어난다.
반대로 프로듀서가 보내는 양이 컨슈머의 데이터 처리량 보다 적으면 컨슈머 랙은 줄어들고 최솟값은 0으로 지연이 없음을 뜻한다.
컨슈머 랙을 모니터링함으로써 컨슈머의 장애 확인 및 파티션 개수를 정하는 데에 참고할 수 있다.
# 컨슈머 랙을 확인하는 방법은 3가지
- 카프카 명령을 사용하여 컨슈머 랙 조회
- 컨슈머 애플리케이션에서 metrics() 메서드 사용
- 외부 모니터링 툴 사용
1. 카프카 명령을 사용하여 컨슈머 랙 조회
$ bin/kafka-consumer-groups.sh --bootstrap-server my-kafka:9092 \
--group test-group --describe
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID
test-group test 0 37 37 0 - - -
2. 컨슈머 metrics() 메서드 사용
for (Map.Entry<MetricName, ? extends Metric> entry : kafkaConsumer.metrics().entrySet()) {
if ("records-lag-max".equals(entry.getKey().name()) |
"records-lag".equals(entry.getKey().name()) |
"records-lag-avg".equals(entry.getKey().name())) {
Metric metric = entry.getValue();
logger.info("{}:{}", entry.getKey().name(), metric.metricValue());
}
}
컨슈머 랙 관련 모니터링 지표는 3가지로 records-lag-max, records-lag, records-lag-avg 이다.
번외로 아래 코드는 요즘 핫한 OpenAPI의 ChatGPT 에서 작성해준 코드인데 문맥은 맞지 않지만 흡사한 형태의 코드로 작성해주었다.
Metrics consumerMetrics = consumer
Map<MetricName, ? extends Metric> metricsMap = consumerMetrics
for (Map.Entry<MetricName, ? extends Metric> entry : metricsMap.entrySet()) {
MetricName metricName = entry
if (metricName.group()
KafkaMetric kafkaMetric = (KafkaMetric) entry
Double consumerLag = (Double) kafkaMetric
System.Metrics;
Metrics consumerMetrics = consumer.metrics();
Map<MetricName, ? extends Metric> metricsMap = consumerMetrics.metrics();
for (Map.Entry<MetricName, ? extends Metric> entry : metricsMap.entrySet()) {
MetricName metricName = entry.getKey();
if (metricName.group().equals("consumer-fetch-manager-metrics") && metricName.name().equals("records-lag-max")) {
KafkaMetric kafkaMetric = (KafkaMetric) entry.getValue();
Double consumerLag = (Double) kafkaMetric.value();
System.out.println("Consumer lag: " + consumerLag);
}
}
뜬금없지만 너무 궁금해서 컨슈머 랙을 조회하는 코드 작성해줘 라고만 했을 뿐인데 뚝딱 만들어주는게 요놈 참 물건일세..
3. 외부 모니터링 툴을 사용하여 컨슈머 랙 조회
컨슈머 랙을 모니터링 하는 가장 최선의 방법은 외부 모니터링 툴 사용하는 것이다.
데이터 독, 컨플루언트 컨트롤 센터와 같은 카프카 클러스터 종합 모니터링 툴을 사용하면 카프카 운영에 필요한 다양한 지표를 모니터링할 수 있다.
모니터링 지표에는 컨슈머 랙도 포함되어 있기 때문에 클러스터 모니터링과 컨슈머 랙을 함께 모니터링하기에 적합하다.
컨슈머 랙 모니터링만을 위한 툴로 오픈소스로 공개되어 있는 버로우(Burrow)가 있다.
3.1 카프카 버로우
버로우는 링크드인에서 개발하여 오픈소스로 공개한 컨슈머 랙 체크 툴로서 REST API를 통해 컨슈머 그룹별로 컨슈머 랙을 확인할 수 있다.
| 요청 메서드 | 호출 경로 | 설명 |
| GET | /burrow/admin | 버로우 헬스 체크 |
| GET | /v3/kafka | 버로우와 연동 중인 카프카 클러스터 리스트 |
| GET | /v3/kafka/{클러스터 이름} | 클러스터 정보 조회 |
| GET | /v3/kafka/{클러스터 이름}/consumer | 클러스터에 존재하는 컨슈머 그룹 리스트 |
| GET | /v3/kafka/{클러스터 이름}/topic | 클러스터에 존재하는 토픽 리스트 |
| GET | /v3/kafka/{클러스터 이름}/consumer/{컨슈머 그룹 이름} | 컨슈머 그룹의 컨슈머 랙, 오프셋 정보 조회 |
| GET | /v3/kafka/{클러스터 이름}/consumer/{컨슈머 그룹 이름/status | 컨슈머 그룹의 파티션 정보, 상태 조회 |
| GET | /v3/kafka/{클러스터 이름}/consumer/{컨슈머 그룹 이름}/lag | 컨슈머 그룹의 파티션 정보, 상태, 컨슈머 랙 조회 |
| DELETE | /v3/kafka/{클러스터 이름}/consumer/{컨슈머 그룹 이름} | 버로우에서 모니터링 중인 컨슈머 그룹 삭제 |
| GET | /v3/kafka/{클러스터 이름}/topic/{토픽 이름} | 토픽 상세 조회 |
버로우는 다수의 카프카 클러스터를 동시에 연결하여 컨슈머 랙을 확인한다.
한 번의 설정으로 다수의 카프카 클러스터 컨슈머 랙을 확인할 수 있다.
버로우는 컨슈머와 파티션의 상태를 단순히 컨슈머 랙의 임계치로 나타내지 않는다.
컨슈머 랙이 특정 시점에 100만이 넘었다고 컨슈머 또는 파티션에 이슈가 있다고 단정 지을 수 없다.
프로듀서가 데이터를 많이 보내면 일시적으로 임계치가 넘어가는 현상이 발생할 수 있기 때문이다.
버로우에선 임계치가 아닌 슬라이딩 윈도우계산을 통해 문제가 생긴 파티션과 컨슈머 상태를 표현한다.
컨슈머 랙의 상태를 표현하는 것을 컨슈머 랙 평가라고 부른다.
랙과 파티션의 오프셋을 슬라이딩 윈도우로 계산하면 상태가 정해진다.
#컨슈머 랙 모니터링 아키텍쳐
- 버로우 : REST API를 통해 컨슈머 랙을 조회할 수 있다.
- 텔레그래프 : 데이터 수집 및 전달에 특화된 툴. 버로우를 조회하여 데이터를 엘라스틱서치에 전달한다.
- 엘라스틱서치 : 컨슈머 랙 정보를 담는 저장소
- 그라파나 : 엘라스틱서치의 정보를 시각화하고 특정 조건에 따라 슬랙 알람을 보낼 수 있는 웹 대스보드 툴
