[Kafka] Consum한 뒤 에러가 나는 경우, 재시도를 막는 방법

⚠️ 경고 ⚠️
본 카테고리, Dev Memo는 필자가 깊게 다루기는 귀찮지만 궁금한 것들을 체험해보고 간단하게 기록을 남기는 공간입니다. 디버깅 노트, 써드파티 라이브러리 사용기, 버전 업데이트, 어이없는 실수, 오탈자 발견 등. 각종 시덥지 않은 내용이 들어 갈 수 있다는 점 참고바랍니다. (우헤헿)

상황 설명

이벤트를 수신(Consume)한 뒤 원하는 로직을 처리하던 중에 에러가 발생할 수 있다. 이후에 다시 메시지를 처리할 수 있도록 재시도를 한다. 필자의 경우 10회동안 반복해서 시도를 했는데 한번 안되는 이상 10번을 해도 같은 상황이었다. 메세지 처리를 반복을 하지 않거나 한번만 반복하도록 하고 싶다. 개발 환경은 스프링 + 코틀린이다.

 

돌고 돌고 반복해서 처리하는 이벤트를 막아보자

 

ListenerContainerCustomizer

단일 Bean이 어플리케이션 컨텍스트에 있는 경우, 바인더에 의해 생성된 리스너 컨테이너는 모든 속성이 설정된 후에 추가로 사용자 정의를 할 수 있다. 그것이 ListenerContainerCustomizer의 역할이다. 여기에서는 에러를 컨트롤하고 싶기에 컨테이너에 에러 설정을 추가하려고 한다. 

 

@Configuration
class KafkaDomainEventConfiguration {

    /**
     * 처리가 실패하면 다시 메시지를 처리하는 횟수 조정
     * max-attempts: 1 <-- 전체 횟수, 즉 재시도 없음
     */
    @Bean
    fun listenerContainerCustomizer(): ListenerContainerCustomizer<AbstractMessageListenerContainer<String?, String?>>? {
        val errorHandler = SeekToCurrentErrorHandler(FixedBackOff(0, 0)) // 재시도 간격 0, 재시도 횟수 0
        return ListenerContainerCustomizer { container: AbstractMessageListenerContainer<String?, String?>, _: String?, _: String? ->
            container.setErrorHandler(errorHandler)
        }
    }

}

 

SeekToCurrentErrorHandler라는 에러 핸들러를 만들고 FixedBackOff로 BackOff를 담는다. BackOff는 작업 재시도 정보를 담은 인터페이스다. 이 설정을 추가함으로 인해 에러 발생시 재시도를 막을 수 있다.

 

 

maxAttempts

여기까지만 설정했을 때, 10번의 처리 요청은 나타나지 않았지만 3번 처리 요청이 나타났다. 기대했던 것과 다르다. Kafka에서 maxAttempts 설정을 추가하면 확실하게 반복을 막을 수 있다. 위 코드의 주석을 참고하고 아래는 yml 파일에 적용한 모습니다. maxAttempts는 최대 수행 횟수이므로 1을 입력하면 재시도를 하지 않고, 2를 입력하면 1회 재시도를 하게된다.

spring:
  cloud:
    stream:
      bindings:
        test-event-in-0:
          destination: test-service.test-event
          content-type: application/json
          group: test-group
          consumer:
            max-attempts: 1

 

참고자료
- Kafka - Spring cloud stream kafka(스프링 클라우드 스트림 카프카)
- Kafka Consumer retry 및 deadletter 처리 방법
- kafka maxAttempts not working and ...
- Spring Cloud Stream Kafka retries 10 times the maxAttempts

.