[MSA] Kafka 컨슈머 Inbox 패턴

시작하며

---

안녕하세요. 개발자 stark입니다.

 

최근 Transactional Outbox Pattern에 대한 대화를 하게 되었는데 제가 아직 이 패턴에 대해 모르는 점이 많다는 것을 깨달았습니다. 그래서 이 궁금증을 해결하기 위해 좀 더 깊이 공부해 봤습니다. 특히 이번 포스팅은 이전에 제가 작성했던 outbox 패턴 포스팅의 내용을 기반으로 이어서 설명드릴 예정입니다. 그러니 아래의 포스팅을 꼭 확인해 주세요!

[MSA] Transactional Outbox Pattern

이전 포스팅에 작성했던 내용을 간략히 설명드리자면 회원가입 http request 요청을 받으면 메서드에서 본인의 작업(회원가입)을 처리하고 이벤트를 발행합니다. 이벤트가 발행되면 리스너가 동작해서 Outbox 테이블에 이벤트 정보를 기록한 후 Kafka로 메시지를 produce 합니다. 그러면 다른 서버의 kafka 리스너들이 이 메시지를 consume 해서 필요한 작업(동기화)을 진행합니다.

 

이 과정에서 스프링 이벤트의 before_commit, after_commit 기능을 활용하였으며 zeroPayload 방식을 적용하여 kafka 메시지의 데이터(payload)에는 id값만 전달하도록 했습니다. 그러나 데이터를 consume 하는 서버가 여러 개 존재할 때는 zeroPayload 방식으로 id만 보내서 http 요청을 하는 것보다는 필요한 데이터를 담아서 보내는 것이 더 적절하다는 생각이 들어서 이번 포스팅에서는 fullPayload로 이전 포스팅의 내용을 재구성한 다음 설명을 이어가겠습니다. (아래와 같이 데이터 처리 부분을 변경하였습니다!)

transactional-outbox-pattern

근데 제가 구성한 Transactional Outbox Pattern에서는 메시지를 발행(produce)하는 서버에 대한 트랜잭션만 보장되어 있습니다. 그렇다면 이 메시지를 소비(consume)하는 서버에서 수신 중에 문제가 발생한다면 어떻게 될까요? 네트워크 문제나 서버 장애는 언제든지 발생할 수 있습니다. 즉, 언제든 Kafka 메시지를 수신하지 못하는 오류가 발생할 수 있다는 것입니다. 그러니 우리는 이런 경우를 예방하기 위해 미리 각 서버의 consume 여부를 파악하는 설계를 해줘야만 합니다.

 

또한 같은 메시지를 여러 구독자(컨슈머)들이 중복으로 여러 번 전달받을 수 있는 가능성도 있습니다. 따라서 컨슈머 측에서도 중복 수신을 대비해야 합니다. 그리고 저는 이것을 inbox 테이블을 사용해서 처리하고자 합니다. 그럼 지금부터 inbox를 적용해 봅시다!

 

 

인박스(Inbox) 테이블의 목적

---

컨슈머 서비스(서버)에서는 인박스 테이블을 두어 수신한 메시지의 처리 이력을 저장하는 패턴을 활용할 수 있습니다. 이 패턴의 주요 목적과 이점은 다음과 같습니다.

 

1. 중복 처리 방지 (Idempotency)

Inbox 테이블에 이미 처리된 메시지의 고유 ID를 기록해 두고, 새 메시지 수신 시 이미 처리된 ID인지 확인합니다. 동일 메시지가 재전달되더라도, 해당 ID가 테이블에 존재하면 중복으로 간주해 처리하지 않음으로써 중복으로 인한 부작용을 막을 수 있습니다. 이를 통해 “한 번만 처리(Exactly-Once)”에 가까운 효과를 얻습니다.

 

2. 트랜잭션 경계 보장 및 원자성

메시지를 처리하면서 Inbox 테이블에 ID를 기록하는 작업을 하나의 DB 트랜잭션으로 수행합니다. 즉, 메시지 처리 로직과 처리완료 표시(Inbox 삽입)를 한 덩어리로 묶어 원자적으로 수행하므로, 일부만 반영되거나 누락되는 일이 없습니다. 데이터베이스의 고유 제약조건(Unique Key)을 이용하여 동일 ID 입력 시 트랜잭션을 롤백함으로써 중복 메시지에 대한 원자적 중단 처리도 가능합니다​.

 

3. 신뢰성 있는 재시도 보장

컨슈머에서 처리 실패 시 메시지를 Ack(확인) 하지 않고 그대로 두면, Kafka는 해당 메시지를 재전송(재시도)합니다. Inbox 패턴을 사용하면 이전 시도에서 부분적으로 완료된 작업이 있더라도 테이블을 통해 이미 처리됨을 인지하고 안전하게 멱등 처리할 수 있습니다. 즉, 재처리 시나리오에서 동일 작업이 두 번 수행되지 않도록 보장하여 시스템 상태의 일관성 유지에 기여할 수 있습니다.

 

4. 메시지 순서 및 처리 속도 제어 (추가 이점)

부가적으로, Inbox 테이블을 활용하면 수신된 이벤트를 테이블에 적재한 후 별도 워커를 통해 내부 처리 순서를 제어할 수 있습니다. 예를 들어 장시간 걸리는 작업의 경우 바로 Ack를 보내고 백그라운드에서 처리하여 Kafka 재전송 타임아웃 등을 유연하게 대처할 수 있고, Inbox 테이블 내에서 메시지 순서를 재조정하거나 병렬 처리를 조율하는 전략도 구현할 수 있습니다. (다만 일반적으로 Kafka는 파티션 단위로 순서를 보장하므로, 특별한 이유가 없다면 컨슈머에서 즉시 처리하고 Ack 하는 방식이 주로 사용됩니다.)

 

 

인박스 패턴 구현 예시 (Kafka 컨슈머 처리 흐름)

---

inbox 테이블 상세 구조

위 다이어그램은 인박스 패턴에서 컨슈머 서비스가 수신한 메시지를 처리하는 전체 흐름을 간단하게 보여줍니다. kafka 컨슈머(리스너)가 메시지를 받으면, Inbox 테이블에 해당 메시지 ID와 구독자(컨슈머) ID를 새로운 행(row)으로 기록합니다. 이 INSERT 연산은 비즈니스 로직에서 수행하는 기타 도메인 데이터(엔티티) 업데이트와 함께 동일한 트랜잭션 내에서 이루어집니다​.

 

만약 Inbox 테이블에 동일한 Message ID가 이미 기록되어 있다면, 기본 키 충돌(Unique constraint)로 INSERT가 실패하고 트랜잭션이 롤백되어 이후 비즈니스 로직이 실행되지 않습니다. 이러한 과정을 통해 중복 메시지에 대한 처리 자체를 차단하고, 하나의 메시지에 대한 처리와 데이터베이스 변경이 원자적으로 일어날 수 있게 보장합니다.

 

 

Kafka 컨슈머에 인박스 패턴 적용하기

---

1. 고유 ID가 있는 메시지 수신

생산자 서비스가 발행한 Kafka 메시지에는 보통 해당 이벤트를 식별할 수 있는 고유 ID(message id)가 포함됩니다 (메시지 payload 필드나 Kafka 헤더에 ID를 전달). 참고로 Outbox 테이블에도 이 고유 메시지 ID를 저장해야 합니다!

{
  "messageId": "msg-001",
  "eventType": "USER_SIGNUP",
  "userId": "user-123",
  "email": "user@example.com",
  "timestamp": "2025-04-06T15:00:00Z"
}

2. DB 트랜잭션 시작 및 중복 확인

kafka 컨슈머(리스너)는 데이터베이스 트랜잭션을 시작한 후, 우선 Inbox 용 처리 이력 테이블(Inbox 테이블)에 새 메시지의 ID를 INSERT 시도합니다. 이 테이블은 (구독자 ID, 메시지 ID)를 복합키로 설정하여, 한 번 처리한 메시지는 다시 삽입 불가하도록 구성합니다.

• 만약 INSERT가 성공하면 해당 메시지는 처음 처리되는 것이므로 계속해서 비즈니스 로직을 수행합니다.
• INSERT 시 기존에 같은 ID가 존재하여 실패하면, 이미 이전에 처리된 중복 메시지이므로 해당 트랜잭션을 abort/rollback 하고 이후 작업을 생략합니다​.
• 이 경우 해당 메시지는 즉시 처리 완료로 간주하고 Ack를 보냄으로써, 중복 메시지를 소비에서 제거합니다​.

@KafkaListener(topics = "user-events", groupId = "user-signup-group")
public void listen(String message, Acknowledgment ack) {
    try {
        // 1. JSON 메시지를 UserSignupEvent 객체로 변환
        UserSignupEvent event = objectMapper.readValue(message, UserSignupEvent.class);

        // 2. 이벤트 처리 (인박스 기록 및 비즈니스 로직 수행)
        userSignupEventService.processUserSignupEvent(event);

        // 3. 모든 처리가 성공하면 Kafka에 메시지 Ack 전송
        ack.acknowledge();
    } catch (Exception e) {
        // 예외 발생 시 로그 기록 및 재처리 로직 구현 (예: Dead Letter Queue 활용)
        e.printStackTrace();
    }
}

kafka 리스너에서는 아래의 서비스를 호출합니다.

@RequiredArgsConstructor
@Service
public class UserSignupEventService {

    private final InboxRecordRepository inboxRecordRepository;
    private final ProfileService profileService; // 유저 프로필 생성 등의 비즈니스 로직 서비스
    
    // 예를 들어, application.properties 에서 consumer.id 값을 주입받음
    @Value("${consumer.id}")
    private String consumerId;

    @Transactional
    public void processUserSignupEvent(UserSignupEvent event) {
        // 1. 중복 체크: 이미 처리된 이벤트인지 확인
        Optional<InboxRecord> existingRecord = inboxRecordRepository.findByMessageId(event.getMessageId());
        if (existingRecord.isPresent()) {
            // 이미 처리된 이벤트이므로 더 이상 진행하지 않음
            return;
        }
        
        // 2. Inbox 테이블에 이벤트 기록 (초기 상태: "PROCESSING")
        InboxRecord record = new InboxRecord();
        record.setConsumerId(consumerId);  // 동적으로 주입된 컨슈머 식별자 (예: 서비스 이름 또는 ID)
        record.setMessageId(event.getMessageId());
        record.setEventType(event.getEventType());
        record.setStatus("PROCESSING");
        
        inboxRecordRepository.save(record);
        
        // 3. 비즈니스 로직 수행: 예를 들어, 유저 프로필 생성
        profileService.createUserProfile(event.getUserId(), event.getEmail());
        
        // 4. 처리 완료 후 Inbox 테이블의 상태를 "SUCCESS"로 업데이트
        record.setStatus("SUCCESS");
        inboxRecordRepository.save(record);  // 업데이트
    }
    
}

3. 비즈니스 로직 수행

위의 코드를 보면 (중복이 아닌 경우에) 메시지 내용에 따른 비즈니스 처리를 진행합니다. 유저 회원가입 이벤트를 기반으로, 다른 시스템 또는 데이터베이스와의 동기화 작업을 진행합니다. 예를 들어, 사용자 정보를 다른 마이크로서비스나 캐시 시스템에 동기화하는 작업이 있을 수 있습니다. 이 로직은 아직 트랜잭션 안에서 실행 중이며, 필요한 경우 여러 DB 변경 작업을 포함할 수 있습니다.

 

4. 트랜잭션 커밋

비즈니스 로직이 성공적으로 완료되면 데이터베이스 트랜잭션을 커밋합니다. 이 시점에 앞서 INSERT 된 Inbox 테이블의 메시지 ID 기록과 도메인 데이터 변경사항이 모두 확정되어 영구 반영됩니다. 만약 비즈니스 로직 중 오류가 발생하면 트랜잭션을 롤백하여 Inbox 테이블에도 기록을 남기지 않고, 이후 Kafka Ack를 보내지 않음으로써 처리를 재시도하도록 합니다​.

 

위의 1~4번 설명은 바로 아래의 다이어그램을 글로 표현한 것입니다.

inbox 패턴 상세 설명

5. 메시지 오프셋 커밋(Ack)

데이터베이스 커밋이 완료되었다면, 컨슈머는 Kafka 브로커에 해당 메시지 처리 완료(Acknowledgement)를 전달합니다. Spring Kafka의 경우 수동 커밋 모드일 때 Acknowledgment.acknowledge()를 호출하거나, 자동 커밋이라면 트랜잭션 종료 시점을 맞추어 커밋하도록 조정합니다. 이로써 Kafka 측에서도 해당 메시지가 소비되었음을 확인하고 더 이상 재전송하지 않습니다​.

 

위 과정으로 컨슈머 서비스는 최소 한 번 전달되는 메시지를 중복 없이 한 번만 처리하도록 합니다. 메시지가 재전달되더라도 Inbox 테이블에 이미 처리된 ID가 존재하므로, 동일한 작업을 다시 수행하지 않고 빠르게 Ack 하여 Kafka 오프셋만 이동시킵니다.

 

 

다이어그램 정리

---

Transactional Outbox Pattern + Inbox 최종 다이어그램입니다.

• 사이즈를 크게 하면 화면에 이상하게 표시가 되어 작게 올립니다. 필요시 확대해서 확인 부탁드립니다 ㅠㅠ

outbox with inbox pattern

이번 포스팅에서 Outbox 패턴에 Inbox 패턴을 추가했기 때문에 이제 메시지를 produce 하는 서버와 consume 하는 서버 모두 이벤트(메시지) 정보를 기록하기 때문에 관련 문제가 발생했을 경우 더 빠르고 확실하게 찾고 재발행할 수 있게 되었습니다.

 

참고로 Outbox table에도 message_id가 추가되었습니다. Outbox와 Inbox가 동일한 message_id를 가지고 있어야 추후 이것을 기반으로 추적이 가능하기 때문입니다. (메시지 발행 전에 id로 유니크한 값을 생성해 주시면 됩니다.) 다음 포스팅에서는 message_id를 활용해서 이벤트를 어떻게 찾아서 재발행하는지도 알아볼 예정이니 기대해 주세요!

 

 

실패 및 재시도 시나리오에서 Inbox의 역할

---

인박스 테이블은 실패 상황과 재시도 시나리오에서 컨슈머의 신뢰성을 크게 향상시킵니다. 몇 가지 가상 시나리오를 통해 그 역할을 살펴보겠습니다.

 

시나리오 1: 컨슈머 처리 중 장애가 발생한 경우 (DB 커밋 전)

컨슈머 서비스가 메시지를 받아 DB 트랜잭션 내에서 Inbox INSERT 및 비즈니스 로직을 수행하던 중, 커밋 전에 장애(예: 애플리케이션 크래시)가 발생했다고 가정해 봅시다. 이 경우 해당 트랜잭션은 커밋되지 않았으므로 Inbox 테이블에 기록이 남지 않고, 도메인 데이터도 변경되지 않을 것입니다. 또한 Kafka에도 Ack를 보내지 못했기 때문에, 메시지는 처리되지 않은 것으로 간주되어 재전송됩니다. 재시도 시 컨슈머는 다시 처음부터 동일 메시지를 처리하게 되고, 이전 시도 여부 데이터가 DB에 없으므로 정상적인 절차로 진행됩니다. 결국 Inbox 패턴 적용 시 실패하면 아무것도 커밋되지 않기 때문에 부분 업데이트나 중복 문제가 남지 않고, 다음 시도에서 정상 처리를 재개할 수 있다는 장점이 있습니다.

 

시나리오 2: DB 커밋 후 Ack 전에 장애가 발생한 경우

컨슈머가 DB 트랜잭션은 성공적으로 커밋했지만 Kafka에 Ack를 보내기 전에 장애가 발생한 경우입니다. 이때 데이터베이스에는 Inbox 테이블에 메시지 ID가 기록되고 비즈니스 로직 결과도 반영된 상태입니다. 그러나 Kafka 측에서는 해당 메시지를 아직 처리 완료로 인지하지 못했으므로, 일정 시간이 지나면 해당 파티션의 컨슈머 세션이 만료되어 다른 인스턴스로 재분배(Rebalance)되거나 재시도가 발생하여 동일 메시지가 다시 전달될 것입니다. 이제 새로운 컨슈머 인스턴스(혹은 재시작된 동일 인스턴스)가 메시지를 받으면, 트랜잭션을 열고 Inbox 테이블에 동일 ID INSERT를 시도합니다. 그러나 이미 이전 처리에서 해당 ID가 INSERT 되어 있으므로 고유키 충돌이 발생하여 중복임을 즉시 감지합니다. 컨슈머는 이를 감지하고 트랜잭션을 롤백한 뒤 곧바로 Ack를 보내 메시지를 소비 완료 처리합니다. 이로써 메시지 중복 처리 없이 Kafka 상의 중복 전달 상황을 안전하게 처리하게 됩니다. 결과적으로 메시지는 한 번만 효과를 미치고, 두 번째 도착했을 때는 무시되고 사라집니다.

 

정리하면, Inbox 패턴을 적용하면 컨슈머 수준의 트랜잭션 관리와 멱등성 보장을 통해 장애 상황에서도 데이터 불일치나 중복 작업을 방지하고, 안정적인 재처리를 지원합니다. 메시지 브로커가 제공하는 최소 한 번 전송 보장(at-least-once delivery) 특성과 결합하여, 이중으로 안전장치를 마련하는 셈이라고 볼 수 있습니다.

 

 

출처 모음

---

https://microservices.io/post/microservices/patterns/2020/10/16/idempotent-consumer.html#:~:text=After%20starting%20the%20database%20transaction%2C,transaction%20and%20acknowledge%20the%20message

Handling duplicate messages using the Idempotent consumer pattern

https://event-driven.io/en/outbox_inbox_patterns_and_delivery_guarantees_explained/#:~:text=successfully%20at%20least%20once,marks%20the%20event%20as%20sent

Outbox, Inbox patterns and delivery guarantees explained - Event-Driven.io

https://softwaremill.com/microservices-101/#:~:text=The%20inbox%20pattern%20is%20quite,just%20removed%20from%20the%20inbox

Microservices 101: Transactional Outbox and Inbox

https://www.lydtechconsulting.com/blog-kafka-idempotent-consumer.html#:~:text=An%20Idempotent%20Consumer%20pattern%20uses,no%20further%20action%20takes%20place

Kafka Idempotent Consumer & Transactional Outbox

서치는 ChatGpt의 도움을 받았습니다.