[SSA] Kafka Consumer가 멱등성을 가질 수 있도록 설계

 

트랜잭션 아웃박스 패턴 도입 이벤트 발행기에서 설계한 것처럼 이벤트를 발행한다면

 

#### 데이터 중복 produce 문제

가 발생할 수 있습니다.

 

이벤트 발행에 성공했지만 발행 상태를 변경하는 트랜잭션이 모종의 이유로 실패하는 경우

1. 발행이 완료되었으나 발행 상태가 미발행 혹은 발행 실패 로 계속 남아있을 수 있습니다.
2. 미발행 혹은 발행 실패로 남아있게 되면, 이벤트 재처리기에 의해 동일한 이벤트가 중복 produce되는 문제가 발생합니다.

이 문제를 해결하기 위해 이벤트를 소비(Consume)하는 로직을 멱등(Idempotent)하게 설계해야 합니다.

 

 

 

# 멱등한 Consumer

1. 이벤트마다 고유한 UUID 필드를 추가
   --> 각 이벤트에 유일한 식별자(UUID)를 부여하여, 동일한 이벤트인지 판별할 수 있도록 한다.
2. 이벤트 처리 이력을 DB에 저장
   --> 이벤트를 Consume하는 곳에서 이벤트 처리, 후 처리한 이벤트를 저장한다.
3. 이미 처리된 이벤트라면 아무 작업도 수행하지 않음
   --> 해당 이벤트 처리 이력이 DB에 존재하면 Consumer는 중복 처리를 방지하기 위해 아무런 작업도 수행하지 않는다.

 

# Consumer 예외 처리

Consumer에서는 DefaultErrorHandler를 통해 예외 처리를 진행할 수 있습니다다.

여기서도 필요하면 재시도를 할 수 있지만, 우선 별도의 재시도는 진행하지 않고 예외 발생 시 실패 이력을 저장하겠습니다.

 

이후 Producer와 마찬가지로, 일정 주기마다 스케줄러를 동작시켜 실패 처리된 메시지를 재처리하도록 할 수 있습니다다.

Dead Letter Queue(DLQ)를 활용할 수도 있지만,

아직 학습이 부족하기 때문에 관리가 어려울 것으로 판단했고,

또한 DB에 저장하는 방식으로 처리가 가능하다 생각해서 우선 DB를 활용하기로 결정했습니다.

 

단, DB 자체에 오류가 발생할 경우 실패 이력을 저장하는 과정이 실패할 수 있습니다.

이러한 경우 로그를 통해 복구할 수 있도록 꼼꼼히 로그를 남기도록 했습니다.

 

이후 특정 토픽에서 별도의 재처리가 필요해지는 경우,

Consumer 내부 로직에서 직접 처리하거나 해당 토픽에 대해서만 재처리를 진행하는 방식으로 설정할 수 있습니다.

 

 

 

--> Kafka 컨슈머 예외 처리: ack() 호출에 따라 메시지가 사라질 수 있다?

## Kafka Consumer에서 예외가 발생했을 때 메시지 Acknowledgment(오프셋 커밋) 처리 방식

Spring Kafka에서 메시지 처리 중 예외가 발생했을 때, Acknowledgment 객체를 어떻게 다루느냐에 따라 메시지의 처리가 달라집니다.

단순히 try-catch로 예외를 잡는 것을 넘어, ack.acknowledge() 호출 여부가 데이터 유실로 이어질 수 있습니다.

 

 

### 두 가지 시나리오: ack() vs No ack()

1. 예외 발생 시에도 ack.acknowledge() 호출: "무조건 처리 완료"

아래 코드처럼 catch 블록 안에서 ack.acknowledge()를 호출하면, 메시지 처리에 실패했음에도 불구하고 Kafka 브로커에는 "처리 완료"라고 보고(오프셋 커밋)합니다.

// 1. 예외가 발생해도 offset을 커밋하는 경우
@KafkaListener(topics = "TEST")
public void consume(String message, Acknowledgment ack) {
    try {
        doSomething(message);
        // 성공했으니 처리했다고 저장한다.
        ack.acknowledge();
    } catch (Exception e) {
        log.error("메시지 처리 실패!", e);
        // 실패했지만, 처리했다고 저장한다.
        ack.acknowledge();
    }
}

• 결과: 브로커는 이 메시지가 성공적으로 처리됐다고 알기 때문에, Consumer가 재시작돼도 절대 이 메시지를 다시 보내주지 않습니다. 즉, 실패한 메시지는 영원히 유실됩니다.

조금 더 엄밀히 말하자면, 토픽 메시지를 보내서 로직이 수행됐든 안 됐든, 메시지 자체는 처리했다고 저장하는 것입니다.

 

 

2. 예외 발생 시 ack.acknowledge() 처리 않음: "일단 보류"

catch 블록에서 ack.acknowledge()를 호출하지 않으면, 해당 메시지의 오프셋은 커밋되지 않습니다.

이는 "이 메시지 처리를 보류하겠다"라고 저장하는 것입니다.

// 2. 예외 발생 시 offset을 커밋하지 않는 경우
@KafkaListener(topics = "TEST")
public void consume(String message, Acknowledgment ack) {
    try {
        doSomething(message);
        // 성공했을 때만 처리했다고 보고한다.
        ack.acknowledge();
    } catch (Exception e) {
        log.error("메시지 처리 실패!", e);
        // 실패 시에는 아무것도 하지 않는다.
    }
}

• 결과: 컨슈머가 재시작되면 이 메시지를 다시 받아 처리할 기회가 생깁니다.
• 주의: 하지만 이 방법도 완벽하지 않습니다. 만약 실패한 100번 메시지의 오프셋은 커밋되지 않았지만, 바로 다음에 들어온 101번 메시지가 성공적으로 처리되어 오프셋이 커밋되었다고 가정해 봅시다. Kafka는 파티션의 커밋된 오프셋을 101로 기억하므로, 컨슈머 재시작 시 102번부터 메시지를 가져옵니다. 결국 실패했던 100번 메시지는 재처리할 기회를 얻지 못하고 유실됩니다.

 

 

현재 프로젝트에는 Consumer 수행 도중 예외 발생 시 ack.acknowledge()를 호출하지 않습니다.

 

또한, 예외 발생 시 별도의 retry를 진행하지 않고, DeadLetter에 저장하도록 구현했습니다.

리밸런싱 등이 발생하는 경우에는 중복 메시지를 consume할 수도 있으나,

이미 처리된 기록(uuid로 조회)이 있다면 생략하도록 RecordFilterStrategy를 적용해 뒀습니다.

 

순서 표로 정리하자면 다음과 같습니다.

단계	수행 주체	핵심 동작	결과 및 상태
1. 최초 수신	KafkaIdempotencyFilter	메시지 ID(uuid=1)의 처리 이력 확인 후 이력을 먼저 저장	처리 이력 없음 -> 통과 (DB에 uuid=1 기록됨)
2. 처리 실패	Consumer Logic	비즈니스 로직 수행 중 예외 발생	작업 중단, 예외 throw
3. 오류 처리	DefaultErrorHandler	재시도 없이 DLQ로 메시지 전송	Offset 미커밋, 메시지는 DLQ에 보관
4. 재수신	KafkaConsumer	리밸런싱 후 동일 메시지(offset=10) 재수신	-
5. 중복 방지	KafkaIdempotencyFilter	메시지 ID(uuid=1)의 처리 이력 재확인	이미 이력이 존재함 -> 필터링(무시)

 

 

 

 

 

 

참고 !

로 카프카에서는 Filter와 Interceptor가 있습니다.

메시지 처리 파이프라인의 서로 다른 단계에서 동작하며, 각각의 역할에  차이가 있습니다.

 

Filter는 컨슈머가 메시지를 처리하기 전에 조기에 폐기할지를 결정하는 '입구 컷' 역할을 하며,

Interceptor는 메시지 처리의 전/후 과정에 깊숙이 개입하여 로깅, 수정, 추적 등 부가적인 작업을 수행하는 '조력자' 역할을 합니다.

 

 

구분	RecordFilterStrategy (Filter) 🚪	RecordInterceptor (Interceptor) 👨‍💼
핵심 목적	메시지 폐기(Discard) 여부 결정	메시지 처리 전/후의 부가 작업 수행
실행 시점	메시지 처리 파이프라인의 가장 앞단	Filter 통과 후, 리스너 실행 직전/직후
반환 값	true (폐기) / false (처리)	ConsumerRecord (수정 가능) 또는 null (폐기)
주요 기능	조건에 따른 메시지 선별 및 제거	메시지 내용 검증, 로깅, 헤더 추가, 성능 측정, 예외 처리
컨텍스트	메시지를 리스너에 전달할 가치가 있는지 판단	리스너의 성공/실패 등 처리 결과까지 관여 가능

 

어떤 메시지에 대해, 어떤 방식으로 중복 처리를 할지 세밀하게 제어하는 것이 필요하다면 Interceptor를 사용하겠으나 현재로서는 처리 선별에만 필요하기 때문에 Filter 방식만 사용했습니다.