Kafka Offset Commit의 중요성

저번 포스팅의 마지막 마무리가 바로 이 에러를 찾지 못했던 것이다.

Kafka 에러를 고치게 된 시점

처음에 특정 Consumer만 consume을 못한다고 했었는데, 이는 당연 잘못된 것이었다!!!

그냥 로그를 좀 더 세세하게 찍고 검토를 더 열심히 했어야 했다!

우선 대략적으로 코드를 보면 아래와 같다.

@KafkaListener(topics = "topic", properties = {
        "spring.kafka.consumer.properties.spring.json.type.mapping=com.github.lsj8367.MessageReq"
    })
public void consumeSomething(final MessageReq req, Acknowledgment acknowledgment) {
    pushService.sendPush(req);
    acknowledgment.acknowledge();
}

일단 회사의 카프카 푸시 로직에서는 자동커밋을 사용하지 않는다.

spring kafka 에서는 enable.auto.commit=true 이라는 설정을 통해 자동으로 일정 시간이 지나면 커밋하게 만들 수 있다.

그 시간은 auto.commit.interval.ms라는 옵션을 통해 설정을 진행할 수 있다.

중복이 생기는 현상

max.poll.records로 15개를 가져온다고 가정한다. (왜냐면 내가 15개씩 가져와서 스레드 15개로 한번에 할당하고 있으니까 15개로 했다)

이제 한번의 컨슘을 통해 이 record들을 15개를 가져왔다.

잘 작업하다가 한 7개 작업을 했다(실질적인 푸시까지 완료) 아직 커밋은 일어나지 않았다. 커밋시간은 조금 더 뒤에 일어나는 상황

여기서

배치 서버에서 파티션 1개로 하니까 너무 느리게 적재되는 것 같아요 병렬처리 하게 파티션을 n개로 바꿀 필요성이 있습니다

라고 하면서 파티션을 늘리는 안이 통과되어 바로 파티션을 늘린다. (참고 - 파티션은 늘릴 수만 있고 줄일 수는 없다)

---

리밸런싱 시작....

이러면 8개가 아직 작업이 남았지만 컨슘이 멈추며 리밸런싱을 시작한다.

어? 커밋된게 없네 해서 이전 15개를 다시 읽어 재처리를 진행하게 된다.

지금 남은 이 8개는 중복이 아니지만, 앞서 작업했던 7개의 메시지는 다시 작업을 수행하게 된다.

이런식으로 중복이 발생한다고 한다.

그럼 넌 뭐가 문제였는데?

나는 애초에 auto commit이 아닌 수동커밋으로 작업을 진행했다.

푸시지만 중복으로 발생되면 안되는 그런 푸시 정보이기에 수동으로 작업을 진행했다.

일단 Spring kafka의 기본값이 수동 커밋인 enable.auto.commit=false 이다.

그리고 ackMode를 설정해줄 수 있는데

ackMode는 뭐냐면 KafkaListener의 offset에 대한 commit을 어떻게 줄것인가를 설정하는 모드이다.

Spring의 기본 AckMode는 BATCH이며, 나는 manual_immidiate로 작업했다.

둘의 차이는 이미지로도 적어놨지만...

• BATCH
  • poll() 메서드로 호출된 레코드가 모두 처리된 이후 커밋
  • 스프링 카프카 컨슈머의 AckMode 기본값
• MANUAL_IMMEDIATE
  • Acknowledgment가 승인되면 즉시 commit

이 두가지의 차이가 있다.

에러 현상은 아무런 오류도, 로그도 찍어지지 않는데, 카프카 Producer 메세지를 발행하면 LAG로 적재되었다.

특정 토픽은 정상적으로 컨슘이 되는게 확인이 된게 너무 억울했었다.

그래서 100개를 연속으로 발행해도 LAG로 메시지가 쌓여서 처리를 못하는 현상이 발생해서 2주를 머리쥐어짰다. 😅😅😅

다른 토픽으로 받아봐도 결과가 계속 같아서 너무 짜증이 더 났었다.

카프카도 새로 설치해달라고 데브옵스분께 요청했고 그래서 이래저래 너무 힘들었다. (운이 좋았던건 dev환경의 k8s설정이 좀 이상하게 되어 다시 구성해야 되는 상황에 요청드렸던 운이 좋았다😱)

에러난 지점은...

@Slf4j
@Service
public class PushService {

    public void sendPush(final MessageReq req, Acknowledgment acknowledgment) {
        try {
            //푸시로직이 작성되어있다...
            log.info("push Success : {}", req);
        } catch (SendFailureException e) {
            log.error("push failure : {}", e.getMessage());
            //DLT 처리
        }
        acknowledgment.acknowledge();  // 에러지점
    }
}

이런식으로 예시로 푸시로직이 있다고 가정하고 이렇게 작성하면 당연히 푸시를 보내고 ack를 통해 offset을 전진시켜주었다.

이게 초반엔 되게 잘 작동했는데 새로운 예외케이스만 모르고 이대로 문제가 없는줄 알았다.

이 부분에서 문제가 되었던건 SendFailureException이 발생하지 않고 NPE나 다른 예외가 뜨게 되면 이게 메세지 처리가 안된다.

그래서 예외가 났으니 offset을 전진 안시켰기 때문에 다시 메시지에 적재 되는것이다.

일정 시간이 지나고 다시 또 똑같은 메시지를 컨슘하는 이 부분이 무한 반복하는 컨슘이 시작되는 것이다.

심지어 로그도 안찍는 현상이다.

지나가며 보던 블로그의 글들을 보며 commit은 항상 필수적으로 해줘야 한다는게 있었는데 난 당연히 commit을 무조건 하는것으로 생각한 나의 실수였다.

그러면 LAG에 고정으로 쌓여있냐? 그건 아니라는 소리다. 수치로는 n개가 적재된거로 보이겠지만 0으로 갔다가 다시 쌓였다가 할것 같은 나의 생각이다.

해결 방법

정말 간단하다. 위에서 봤을때 문제점이 무엇이었는가?

바로 offset을 무조건 전진시켜주면 끝난다!!

@Slf4j
@Service
public class PushService {

    public void sendPush(final MessageReq req, Acknowledgment acknowledgment) {
        try {
            //푸시로직이 작성되어있다...
            log.info("push Success : {}", req);
        } catch (SendFailureException e) {
            log.error("push failure : {}", e.getMessage());
            //DLT 처리
        } catch (Exception e) {
            log.error("알 수 없는 오류 발생 : {}", e.getMessage());
        }finally {
            acknowledgment.acknowledge();  // 에러지점
        }
    }
}

이렇게 무조건 commit 처리하게 해주고 저런 예외가 터졌을 경우에는 DLT (Dead Letter Topic)에 적재해주고 후속 처리를 진행하면 되겠다.

정리

무턱대고 구현할 때 좀 더 꼼꼼하게 작성하고 좀 더 테스트를 빡세게 하고 테스트 코드도 좀 더 케이스를 많이 짜려고 노력을 좀 해야겠다.

그래도 이걸 2주동안 붙잡고 있던게 결국 이런 오류들은 어려운 오류가 아닌데에서 나오는 기간이다.

아무튼 기초를 좀 더 탄탄히 하고 항상 꼼꼼하게 하자 🔥🔥🔥🔥🔥