(Kafka) 유효하지 않은 레코드가 토픽에 들어와 카프카 어플리케이션 장애가 났을 때

상황

 특정 토픽을 Consume 어플리케이션에서 장애가 난 상황. 이로 인해 Lag이 천정부지로 치솟고 있었다.

 또한 이 토픽을 사용하는 kafka to s3 커넥터에서도 장애가 발생하였다.

 

원인

 해당 토픽을 사용하는 어플리케이션들은 해당 토픽의 Value를 Json으로 기대하여 Json 파싱하고 있었다. 그런데 유효하지 않은 데이터가 들어가서, Json Parsing 에러를 발생시킨 것.

 

해결 방안

 우선 장애가 발생한 파티션을 찾아야했다. 장애가 발생한 파티션을 찾기 위해서 쉘 스크립트를 만들어 돌려서, 장애 발생 토픽을 찾아냈다.

BROKER="브로커 호스트"
GROUP="장애가 발생한 토픽을 사용하는 컨슈머 그룹 명"
TOPIC="토픽 명"
MAX_MESSAGES=15 (콘솔에서 보여줄 메세지 갯수)

/opt/kafka/bin/kafka-consumer-groups.sh --describe --bootstrap-server $BROKER --group $GROUP | grep labeled | while read -r line; do

    partition=$(echo $line | awk '{print $3}')
    offset=$(echo $line | awk '{print $4}')

    # Consume messages from the topic based on the extracted partition and offset
    echo "Consuming from partition $partition at offset $offset"
    /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server $BROKER \
      --topic $TOPIC \
      --partition $partition \
      --offset $offset \
      --property print.key=true \
      --property key.separator=":" \
      --max-messages $MAX_MESSAGES | grep 'as'
done

 

 즉 장애가 난 토픽은 장애를 발생시키는 레코드의 value 이전에 멈추어 있을 것이고, 이것을 하나씩 돌려보면서 식별하는 것이다.

 

 이제 교체를 위해서는 먼저 해당 topic 을 사용하는 어플리케이션을 종료하여야 한다. 내가 운영중인 어플리케이션은 쿠버네티스상에 띄워져 있으므로 쿠버네티스 스케일 명령어로 replica 갯수를 0으로 줄여 어플리케이션을 잠깐 종료해두었다.

 

kubectl scale deployment 스테이트풀셋 이름 --replicas=0

 

이후에 offset을 현재 메세지 오프셋 + 문제의 메세지 갯수만큼 건너 띄도록 명령어를 날려주었다.

 

./kafka-consumer-groups.sh --bootstrap-server 호스트주소:포트
--group 컨슈머그룹 이름
--topic 토픽 이름:파티션 이름(주의, 파티션 이름을 쓰지 않으면 모든 파티션에 다 적용된다)
--reset-offsets
--to-offset 현재 오프셋 + 문제의 메세지 갯수
--execute

 

 이후 다시 어플리케이션을 기동해주었다. 다시는 장애를 경험하고 싶지 않다...

 

그 외

 커넥터에서도 파싱 오류가 발생하였는데, 커넥터에서 errors.tolerance: all 으로 설정하면, 이러한 단건 에러는 넘어가는 것을 확인하여 해당 설정을 추가해주었다.

 

 

Kafka Connect Deep Dive – Error Handling and Dead Letter Queues | Confluent