컬리의 입고 시스템이 외부 인입 데이터를 안전하게 동기화하는 방법

4 weeks ago 3

아웃박스 패턴과 재시도 토픽으로 외부 채널 입고 정보를 안전하게 동기화하기

이창섭 게시 날짜: 2026.01.12.

들어가며
기본 시스템 구성
아웃박스 패턴
재시도 토픽
맺으며

안녕하세요. SCM프로덕트개발의 이창섭입니다. 2026년 컬리의 첫 번째 기술 블로그를 제가 작성하게 되는군요. 모두 새해 복 많이 받으세요.

2025년 한 해는 컬리에게도, SCM프로덕트에게도 많은 변화가 있었던 한 해였습니다.

기존의 컬리는 1P(First-Party, 직매입) 모델을 중심으로, 자사가 큐레이팅한 상품을 직접 발주하여 물류센터에 입고하여 직접 판매하는 사업이 주류였습니다. 그렇다는 것은 즉, 입고 예정 정보 대부분이 자사의 발주 데이터에 의해 생성되므로, 외부로부터 입고 예정 정보가 수신될 일은 거의 없었다는 것이죠.

그러나 이제는 컬리의 사업이 확장되면서, 자사의 경쟁력 있는 물류 인프라를 외부 파트너사에게도 제공하는 3PL(Third-Party Logistics, 제3자 물류) 사업을 더욱 공격적으로 추진하게 되고, 그에 따라 이제는 입고 예정 정보가 내부에서만 생성되는 것이 아니라, 외부의 여러 채널에서 다양한 규격과 형태로 수신되게 되었습니다.

저희 SCM프로덕트의 입고 파트는, 이 외부 채널에서 생성되는 입고 예정 정보를 가장 안전하고 정확하게 동기화하는 방법에 대해 고민을 했고, 그 결과로 저희 프로덕트에 적용한 아웃박스 패턴과 재시도 토픽에 대한 얘기를 써볼까 합니다.

우선, 외부 채널의 입고 예정 정보를 가져와 동기화하는 시스템의 기본 골자는 아래와 같습니다.

외부 채널에서 컬리에게로 데이터를 전송하는 방법은 크게 두 가지가 있을 것입니다.

컬리가 제공하는 “입고 예정 정보 생성 API” 를 파트너사가 직접 호출하는 방식
파트너사가 제공하는 “입고 예정 정보 조회 API” 를 컬리가 주기적으로 조회해서 최신화하는 방식

두 방식 모두, 이후 스텝은

인터페이스 용도로 사용하는 DB 테이블에 데이터를 적재하고
카프카 메시지를 발행하여
카프카 리스너에서 컬리의 입고 예정 정보 규격에 맞게끔 데이터를 변환하여 최종적으로 입고 예정 정보를 생성한다.

로 이루어집니다.

카프카 메시지 발행에 실패하면 정상적으로 입고 요청 정보를 수신할 수 없으므로, 인터페이스 DB 를 롤백 시키고 에러 처리를 해야 합니다. 이를 위해 1번과 2번 스텝은 하나의 트랜잭션으로 묶었습니다.

그런데, 이 시나리오에선 쉽게 예상해 볼 수 있는 실패 지점이 두 군데가 있습니다.

바로 위와 같이, 프로듀스 패턴에서 카프카 메시지는 발행이 되었는데, 네트워크나 DB Server 의 상태 이슈 등으로 인해 Commit 이 실패할 경우, 인터페이스 테이블과 카프카 메시지 간의 원자성이 깨지게 됩니다.

또, 컨슈머에서도 메시지를 읽어와 변환하는 데까지는 성공했으나, 입고 예정 정보를 생성하기 위한 다른 기반 정보가 아직 생성되지 않았거나 하는 등의 이유로 입고 예정 정보를 생성하는데 실패할 수도 있습니다. 이해가 쉽도록 조금 구체적인 예를 들어보자면, 아직 파트너 정보가 등록되지 않은 파트너사로부터 입고 예정 정보가 전송되었다든지, 파트너사에는 존재하지만 아직 컬리로 전송되지 않은 상품 정보를 기반으로 생성된 입고 예정 정보가 전송되었다든지 하는 경우 등이 있을 수 있습니다.

이 경우는 파트너사 정보를 담당하는 서버 혹은 상품 정보를 담당하는 서버가 정보를 동기화하는데 지연이 있었거나, 혹은 담당자의 단순 실수로 인한 프로세스 순서 뒤바뀜일 수 있습니다. (단, 이 프로세스 순서는 컬리에서의 제약이기 때문에, 외부 채널의 시스템에 이 순서를 강요할 수는 없습니다.)

즉, 얼마가 걸릴지 정확히는 알 수 없지만 일정 시간이 지나고 다시 재시도 하면 정상적으로 처리될 가능성이 아주 큽니다. (Poison Pill 과 같은 이슈에 대해서는 보장됨을 전제로 하는데, 이는 메시지 발행부와 수신부 모두 컬리의 서비스 내에 종속되어 있기 때문입니다.)

그렇다면, 이 플로우가 실패했을 때, 운영자의 개입 없이 시스템이 스스로 자가 치유하기 위해선 어떤 식으로 보완을 하면 될까요? 우선 동기화할 데이터를 받아와서 인터페이스 테이블에 적재하고, 카프카 메시지를 발행하던 발행부를 먼저 보겠습니다.

아웃박스 패턴이란?

아웃박스 패턴은 “DB 쓰기”와 “메시지 발행” 두 작업의 원자성을 보장하는 아키텍처입니다. 원자성이란, 쉽게 말해 “모두 성공일 때만 성공이고, 하나라도 실패한다면 모두 실패한다(All or Nothing)” 라는 특성입니다.

위 시스템에 이 패턴을 도입해 보자면, 아래와 같이 됩니다.

수신한 데이터를 인터페이스용 테이블에 적재합니다.
카프카 메시지를 바로 발행하는 것이 아니라, 발행할 메시지 내용을 별도의 아웃박스 테이블에 적재만 해둡니다.
그리고 아웃박스 테이블을 읽어올 별도의 폴러(Poller)를 두어, 메시지가 아직 발행되지 않은 데이터를 주기적으로 읽어와 카프카 메시지 발행 처리를 합니다.
카프카 메시지가 정상적으로 발행이 되었다면 아웃박스 테이블에 중복으로 발행되지 않게끔 마킹 처리를 합니다.

이를 그림으로 표현해 보면 아래와 같습니다.

하지만, 이렇게 아웃박스 테이블을 별도 만들어 관리하고, 폴러(Poller) 시스템도 별도로 준비해야 하는 등, 생각보다 신경 써야 할 것도, 구현해야 할 내용도 만만치 않습니다. 이 모든 것을 직접 구현하고 관리할 바에는 ‘그냥 에러가 날 때마다 알림이 오게 하고 수동으로 처리할까…’ 하는 생각도 듭니다.

Namastack Outbox for Spring Boot

바로 이런 고민을 하고 있는 우리를 위해 얼마 전(2025년 10월), Namastack Outbox for Spring Boot 라는 라이브러리(https://outbox.namastack.io)가 릴리즈 됩니다. 이는 아웃박스 패턴의 구현을 간편하게 도와주는 경량 라이브러리로, 간편하게 도입하고 사용하기에 썩 괜찮아 보여 채택하게 됩니다.

이 라이브러리가 해결해 주는 영역을 위 그림에서 표시해 보면 아래와 같습니다.

보시다시피 아웃박스 테이블 관리, 폴링 스케줄링, 재시도 횟수/상태 관리 등을 라이브러리에 위임할 수 있어, 우리가 직접 관리해야 할 범위가 크게 줄어듭니다.

그럼 이제, 라이브러리를 어떻게 추가하고, 우리의 소스코드를 어떻게 수정해야 하는지 코드 레벨로 한 번 살펴봅시다.

테이블 생성

우선, 아웃박스 테이블을 수동으로 생성해 줍니다. outbox.schema-initialization.enabled 라는 설정값을 true 로 선언하여 자동으로 테이블이 생성되게끔 할 수도 있지만, 대부분의 회사가 애플리케이션에서 사용되는 DB 유저에겐 DDL 권한을 주지 않습니다. 컬리도 마찬가지고요.

create table outbox_record ( id varchar(255) not null primary key, status varchar(20) not null, aggregate_id varchar(255) not null, event_type varchar(255) not null, payload text not null, created_at datetime(6) not null, completed_at datetime(6) null, retry_count int not null, next_retry_at datetime(6) not null, partition_no int not null ); create index idx_outbox_record_aggregate_completed_created on outbox_record (aggregate_id, completed_at, created_at); create index idx_outbox_record_aggregate_created on outbox_record (aggregate_id, created_at); create index idx_outbox_record_partition_status_retry on outbox_record (partition_no, status, next_retry_at); create index idx_outbox_record_status on outbox_record (status); create index idx_outbox_record_status_retry on outbox_record (status, next_retry_at);

create table outbox_instance ( instance_id varchar(255) not null primary key, hostname varchar(255) not null, port int not null, status varchar(50) not null, started_at datetime(6) not null, last_heartbeat datetime(6) not null, created_at datetime(6) not null, updated_at datetime(6) not null ); create index idx_outbox_instance_last_heartbeat on outbox_instance (last_heartbeat); create index idx_outbox_instance_status on outbox_instance (status); create index idx_outbox_instance_status_heartbeat on outbox_instance (status, last_heartbeat);

주) 위 내용은 namastack-outbox 0.3.0 버전을 기준으로 작성되었습니다. 0.2.0 버전까지는 outbox_record 테이블의 partition_no 컬럼의 이름이 partition 이었으므로, 컬럼명을 다르게 생성해야 하며, 특히 MySQL 등을 사용 시엔 partition 이 예약어와 중복되므로 JPA 사용 시 백틱(`) 사용 등에 주의가 필요합니다.

소스 코드

우선, 아래와 같이 build.gradle 에 의존성을 추가해 줍니다.

이 글을 작성하는 현재(2025-12-26)는 namastack-outbox 0.4.1 버전까지 릴리즈가 되었습니다만, 0.4.x 버전부터는 SpringBoot 4 이상의 버전대와만 호환이 되므로, 본 글에서는 0.3.0 버전을 기준으로 작성합니다.

api("io.namastack:namastack-outbox-starter-jpa:0.3.0")

그리고, application.yml 에 아래와 같은 outbox 설정값들을 추가합니다.

outbox: # 폴링 주기 poll-interval: 2000 # 아웃박스 테이블을 폴링하는 주기(ms) (기본값: 2000) # 배치 크기 batch-size: 10 # 한 번의 배치 작업에서 처리할 최대 aggregate ID 개수 (기본값: 10) # 처리 동작 설정 processing: stop-on-first-failure: true # aggregate 내에서 이벤트 하나가 실패했을 때, 남은 이벤트 처리를 중단할지 여부 (기본값: true) publish-after-save: true # 이벤트를 아웃박스에 저장한 직후 리스너에게 즉시 발행할지 여부 (기본값: true) delete-completed-records: false # 처리가 완료된 아웃박스 레코드를 DB 에서 삭제할지 여부 (기본값: false) executor-core-pool-size: 4 # 스레드풀의 기본 스레드 개수 (기본값: 4) executor-max-pool-size: 8 # 스레드풀의 최대 스레드 개수 (기본값: 8) # 인스턴스 조정 및 파티션 관리 instance: graceful-shutdown-timeout-seconds: 15 # 안전한 종료(Graceful Shutdown)를 위해 대기하는 시간(초) (기본값: 15) stale-instance-timeout-seconds: 30 # 특정 인스턴스가 응답이 없어 '비정상' 상태로 간주하기까지의 시간(초) (기본값: 30) heartbeat-interval-seconds: 5 # 인스턴스가 Heartbeat 를 보내는 시간 간격(초) (기본값: 5) new-instance-detection-interval-seconds: 10 # 새로운 인스턴스 등장을 감지하는 시간 간격(초) (기본값: 10) # 재처리 설정 retry: max-retries: 3 # 실패한 이벤트의 최대 재처리 횟수 (기본값: 3) policy: "exponential" # 재처리 전략: fixed(고정 지연), exponential(지수 백오프), jittered(지터 적용) (기본값: exponential) # 지수 백오프 전략 설정 exponential: initial-delay: 2000 # 첫 번째 재처리 대기 시간(ms) (기본값: 2000) max-delay: 60000 # 최대 대기 시간(ms) (기본값: 60000) multiplier: 2.0 # 재처리간 대기 시간 배수 (기본값: 2.0) # 고정 지연 전략 설정 fixed: delay: 5000 # 재처리 대기 시간(ms) (기본값: 5000) # 지터(Jittered) 재처리 설정 (재처리 대기 시간에 무작위성을 더해 충돌 방지) jittered: base-policy: exponential # 지터를 적용할 기본 전략 (기본값: exponential) jitter: 500 # 기본 전략 대기 시간에 더해질 최대 무작위 시간(ms) (기본값: 500)

공식 문서에서 제공한 가이드에서 특별히 수정할 내용은 없으나, DDL 을 직접 수행하여 테이블을 생성하였으므로 outbox.schema-initialization.enabled 항목만 빼두었습니다.

이어서, namastack-outbox 에서 필요로 하는 Bean 을 등록할 차례입니다. namastack-outbox 는 내부적으로 Clock 객체를 사용하고 있으므로, 이 객체를 Bean 으로 등록해 줘야 합니다. 아래와 같이 클래스 파일을 하나 작성합니다.

@Configuration public class OutboxConfiguration { @Bean public Clock clock() { return Clock.systemUTC(); } }

그리고 마지막으로, 메인 클래스에 @EnableOutbox 어노테이션을 하나 추가해 줍니다.

@SpringBootApplication @EnableOutbox public class InboundApiApplication { public static void main(String[] args) { SpringApplication.run(InboundApiApplication.class, args); } }

이로써 namastack-outbox 라이브러리를 사용할 환경 설정은 모두 완료되었습니다. 참 간단하죠?

서비스 로직: Producer

이제는 인터페이스 테이블에 적재하고, 카프카 메시지를 발행하던 비즈니스 구현부의 로직을 수정할 차례입니다.

우선, 위와 같은 로직이 있었다고 가정합시다.

외부 API 를 조회하여 interface_master 테이블 및 interface_detail 테이블에 저장한 후, 정의해둔 Kafka 메시지 규격에 맞게끔 변환하여 발행하는 로직입니다.

위에서 말씀드린 대로, send 메서드가 수행되어 Kafka 메시지는 정상 발행 되었으나, @Transactional 로 감싸여 있는 process 메서드의 수행이 종료되면서 DB Commit 이 실제로 수행될 때, 모종의 이슈로 인해 Commit 이 실패한다면, save 와 saveAll 에 수행된 내용들은 모두 롤백이 될 겁니다.

그러면 인터페이스 DB 에는 데이터가 없으나, 카프카 메시지는 발행되어버리는 참사를 초래하죠.

이를 아래와 같이 아웃박스 패턴으로 변경합니다.

@Service public class ExternalInboundExpectationSaveService implements ExternalInboundExpectationSaveUseCase { private final InboundExpectationInterfaceMasterPort inboundExpectationInterfaceMasterPort; private final InboundExpectationInterfaceDetailPort inboundExpectationInterfaceDetailPort; // private final KafkaProducePort kafkaProducePort; // 추가된 의존성 private final OutboxRecordRepository outboxRecordRepository; private final ObjectMapper objectMapper; private final Clock clock; public ExternalInboundExpectationSaveService( InboundExpectationInterfaceMasterPort inboundExpectationInterfaceMasterPort, InboundExpectationInterfaceDetailPort inboundExpectationInterfaceDetailPort, // KafkaProducePort kafkaProducePort, // 추가된 의존성 OutboxRecordRepository outboxRecordRepository, ObjectMapper objectMapper, Clock clock ) { this.inboundExpectationInterfaceMasterPort = inboundExpectationInterfaceMasterPort; this.inboundExpectationInterfaceDetailPort = inboundExpectationInterfaceDetailPort; // this.kafkaProducePort = kafkaProducePort; // 추가된 의존성 this.outboxRecordRepository = outboxRecordRepository; this.objectMapper = objectMapper; this.clock = clock; } @Override @Transactional public void process(Request request) { inboundExpectationInterfaceMasterPort.save(request.toMaster()); inboundExpectationInterfaceDetailPort.saveAll(request.toDetail()); InboundExpectationMessage message = InboundExpectationMessage.from(request); // 카프카 메시지를 직접 발행하지 않고, outbox_record 테이블에 적재만 함 try { OutboxRecord outboxRecord = new OutboxRecord.Builder() .aggregateId(message.inboundId()) .eventType("INBOUND_EXPECTATION_SAVE") .payload(objectMapper.writeValueAsString(message)) .build(clock); outboxRecordRepository.save(outboxRecord); } catch (Exception e) { throw new InboundServiceException("Failed to serialize Inbound Expectation Created Event", e); } } }

이렇게만 해주면 서비스 로직의 변경은 끝입니다. (저희는 조금 더 공통화 과정을 거쳐 아웃박스 또한 하나의 port 로 구성했지만, 여기서는 쉬운 이해를 위해 서비스 레이어 내에서 펼쳐 보았습니다.)

OutboxRecordRepository 클래스는 namastack-outbox 라이브러리에 의해 자동으로 추가되었을 것이고, ObjectMapper 클래스는 Spring 기본 라이브러리에 의해, 그리고 Clock 클래스는 아까 우리가 직접 Bean 에 등록해 주었습니다.

서비스 로직 내에서 직접 카프카 메시지를 발행할 것이 아니기 때문에, 기존에 주입받고 있었던 KafkaProducePort 는 이제 더 이상 필요 없게 됩니다.

서비스 로직: Poller

그럼 이제, 이 outbox_record 에 저장된 데이터를 폴링(Polling)하여 실제 카프카 메시지를 발행할 부분을 구현할 시간입니다. 데이터 폴링은 namastack-outbox 라이브러리가 자동으로 해주는데, @EnableOutbox 를 추가했던 것과 마찬가지로, 메인 클래스에 @EnableScheduling 어노테이션만 추가로 하나 더 붙여주면 됩니다.

저는 폴러(Poller) 역할은 위와 동일한 인스턴스(InboundApiApplication)가 아닌, 별도로 띄우고 있는 배치 인스턴스(InboundBatchApplication)에 위임하기 위해 아래와 같이 다른 인스턴스에 추가했습니다.

@SpringBootApplication @EnableBatchProcessing @EnableScheduling @EnableOutbox public class InboundBatchApplication { public static void main(String[] args) { SpringApplication.run(InboundBatchApplication.class, args); } }

그리고 마지막으로, OutboxRecordProcessor 인터페이스를 구현하여 실제 메시지를 발행할 로직을 구현해 주면 됩니다.

@Component @Slf4j public class KafkaProduceOutboxProcessor implements OutboxRecordProcessor { private final ObjectMapper objectMapper; private final KafkaProducePort kafkaProducePort; public KafkaProduceOutboxProcessor(ObjectMapper objectMapper, KafkaProducePort kafkaProducePort) { this.objectMapper = objectMapper; this.kafkaProducePort = kafkaProducePort; } @Override public void process(@NonNull OutboxRecord outboxRecord) { switch (outboxRecord.getEventType()) { case "INBOUND_EXPECTATION_SAVE": this.sendInboundExpectationMessage(outboxRecord); break; default: log.warn("Unsupported OutboxRecord event type: {}", outboxRecord.getEventType()); } } private void sendInboundExpectationMessage(OutboxRecord outboxRecord) { try { InboundExpectationMessage message = objectMapper.readValue(outboxRecord.getPayload(), InboundExpectationMessage.class); kafkaProducePort.send(message); } catch (Exception e) { log.error( "Failed to process OutboxRecord for InboundExpectationMessage. outboxId: {}, error: {}", outboxRecord.getId(), e.getMessage(), e ); } } }

위에서 OutboxRecordRepository 를 통해 데이터를 적재할 때, eventType 을 INBOUND_EXPECTATION_SAVE 로 지정했던 것을 기억하시나요? 여기서는 OutboxRecordProcessor 인터페이스의 process 메서드를 오버라이딩 하여 eventType 이 INBOUND_EXPECTATION_SAVE 일 때, 카프카로 메시지를 발행하는 로직을 구현해 두었습니다.

이렇게만 해두면, poll-interval 주기(기본 2초)마다 한 번씩 outbox_record 테이블을 읽어와, 아직 처리되지 않은 레코드를 처리하고, 만약 실패했을 경우엔 설정값에 따른 재시도 횟수만큼 재시도를 실행하고, 성공하면 더 이상 polling 되지 않도록 마킹하는 처리까지 모두 라이브러리 내부에서 알아서 처리해 주게 됩니다.

단, Kafka 메시지 발행에 성공하고, 해당 아웃박스 레코드를 완료 상태로 마킹하는 사이에 인스턴스가 다운된다거나 하는 상황이 생길 수 있습니다. 이런 경우, 인스턴스가 다시 구동되게 되면 같은 메시지가 두 번 발행될 수 있는 이론적 위험이 도사리고 있습니다. 이를 인지하고 Kafka 리스너에서 멱등성을 보장할 수 있는 설계가 미리 마련되어야 합니다.

이로써 Kafka 메시지를 발행하는 부분을 아웃박스 패턴으로 안전하게 보완하는 작업은 완료되었습니다. 이제는 이 메시지를 수신하는 부분을 재시도 토픽을 적용해 보완하는 작업을 이어가 봅시다.

일반적으로 Kafka 에서 Consumer 가 메시지를 처리하다 실패하면, 성공할 때까지 계속 시도하는 과정에서 뒤에 대기 중인 메시지들이 처리되지 못하는 Blocking 현상이 발생합니다. 이를 방지하기 위해 저희 컨슈머 프로젝트에는 메시지를 컨슘하던 중 에러가 발생하면

슬랙을 통해 개발자들에 알림을 발송하고
비즈니스 로직은 실패한 채로 메시지를 커밋 처리

하는 ErrorHandler 가 탑재되어 있습니다. 아마 대다수의 프로젝트들에 이와 비슷한 장치들이 있을 것으로 생각합니다.

그런데 이런 방식으로 구동되고 있을 경우에, 위에서 설명한 것과 같이

아직은 이 메시지가 처리될 수 없는 상태이지만,
얼마가 걸릴지 정확히 알 수는 없으나, 그렇게 길지는 않은 어느 정도의 시간이 지나고 나서는 정상적으로 처리가 될 것으로 예상이 되는

메시지일 경우엔 어떻게 할까요?

현재와 같은 시스템이라면, 실패한 메시지를 개발자가 따로 모아서 체크해놓고 있다가, 메시지를 정상 수신할 수 있는 상태가 되면 인위적으로 메시지를 다시 발송 처리를 해줘야 합니다. 일반적으로 운영 환경에서는 메시지를 수동 발행할 수 있는 권한도 따로 나누어 놓기 때문에, 프로세스를 통해 승인을 득하거나 권한자에게 따로 연락해야 할 수도 있습니다.

그리고 24시간 열려 있는 외부 채널에서는 언제 새로운 데이터가 들어올지 알 수 없으니, 이러한 현상은 모두가 잠들어 있는 새벽에도 충분히 발생할 수 있습니다.

재시도 토픽이란?

이러한 경우에 재시도 토픽을 사용하면 아주 효과적입니다. 재시도 토픽의 작동 원리는 아래와 같이 간략하게 설명할 수 있습니다.

메인 토픽(main-topic) 컨슘: 메시지 처리 중 예외 발생.
재시도 토픽 전송: 실패한 메시지를 main-topic-retry 와 같은 이름의 재시도 전용 토픽으로 보냅니다.
지연 후 재시도: 각 재시도 토픽은 설정된 지연 시간(Backoff) 만큼 기다렸다가 메시지를 다시 컨슘합니다.
DLT: 지정된 횟수만큼 재시도해도 실패하면 최종적으로 DLT(Dead Letter Topic)로 보내 관리자가 확인하게 합니다.

즉, 토픽 자체를 분리하여 재시도 토픽으로 메시지를 보내서 거기에서 재시도가 진행되게 함으로써, 뒤에 쌓여있는 메시지들을 블락하지 않도록 운영할 수 있는 기법입니다.

이쯤에서, 아까 위에서 말씀드렸던 저희의 상황을 다시 한번 살펴보겠습니다.

즉, 얼마가 걸릴지 정확히는 알 수 없지만 일정 시간이 지나고 다시 재시도 하면 정상적으로 처리될 가능성이 아주 큽니다.

보통 입고 예정 정보는 1~2일쯤 전에 미리 생성을 합니다. 즉, 입고 예정 정보가 생성되고 나서 실제 상품이 물류 센터에 도착하는 데까지 걸리는 이 1~2일의 시간, 이 시간 텀이 이 메시지가 처리되어야 할 데드라인이라고 볼 수 있습니다.

따라서 저희는, 실패한 메시지를 재시도 토픽으로 보내 10분 간격으로 최대 24시간 동안, 즉 10분 간격으로 총 144회 재시도 처리를 시키고자 합니다.

이를 그림으로 표현해 보면 아래와 같습니다.

네. 역시나 생각한 것보다는 신경 써야 할 부분이 꽤 보입니다. 재시도 토픽을 별도로 구성해야 하고, 10분 지연을 어떻게 구현할지도 신경 써야 하며, 데드레터 토픽을 또 별도로 구성해 144회 재시도 횟수를 모두 소진하고 나면 메시지를 데드레터 토픽으로 보내야 합니다. 이러한 메커니즘을 구성하는 것이 만만치는 않아 보입니다.

Spring Kafka - RetryableTopic

역시나 이런 저희의 고민을 타파해 줄, Spring 에서 공식적으로 제공해 주는 강력한 기능이 있습니다. 바로 Spring Kafka 의 RetryableTopic 입니다. 이 또한 위 그림에서 어떤 부분까지 커버해 주는지 영역을 표시해 보겠습니다.

Spring Kafka 의 RetryableTopic 이 적용된 컨슘 흐름도

크~ 우리가 구현을 고민하던 대부분의 영역을 Spring Kafka 가 기능으로 제공해 주고 있습니다. 우리는 많은 것을 신경 쓸 필요 없이, 기존에 작성해 두었던 컨슈머 로직을 그대로 사용할 수 있어 보입니다.

그럼, 이것도 한 번 코드 레벨까지 살펴봅시다.

소스 코드

Spring Kafka 의 RetryableTopic 은 어노테이션으로 아주 간편하게 설정이 가능합니다.

@RetryableTopic( attempts = "145", backoff = @Backoff(delay = 600000L), sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC, kafkaTemplate = "retryKafkaTemplate" )

기존의 Kafka Listener 에 위와 같이 어노테이션만 하나 추가해 주면 됩니다. 허무할 정도로 간단하죠?

물론, 잡다하지만 신경 써야 할 부분은 있었습니다.

우선 가장 먼저, 이 재시도 전략은 메인 토픽에서의 횟수를 포함하여 총 컨슘 시도 횟수가 최대 145회나 됩니다. 이 때, sameIntervalTopicReuseStrategy 설정을 따로 지정해 주지 않는다면, 기본값인 MULTIPLE_TOPICS 로 설정됩니다. 이럴 경우 재시도 토픽이 144개나 추가로 생성되기 때문에 카프카 클러스터에 부하를 줄 수 있는 여지가 있을뿐더러, 실패한 메시지가 현재 몇 번째 재시도 되고 있는지 확인해 보기 위해선 수많은 재시도 토픽을 일일이 찾아봐야 하게 됩니다. 따라서 저흰 SINGLE_TOPIC 으로 설정하여 하나의 재시도 토픽만을 운용하는 전략을 선택했습니다.

또, 저희 프로젝트의 경우엔 카프카를 발행할 때, 객체를 그대로 넘겨서 자동으로 직렬화되게끔 아래와 같은 Kafka Template 을 구성하고 있었습니다.

@Bean public KafkaTemplate<String, Object> kafkaTemplate() { Map<String, Object> configs = new HashMap<>(); configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers); configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(configs)); }

하지만 메인 토픽에서 재시도 토픽으로 메시지를 발행할 때도 같은 템플릿을 사용하게 된다면, 이미 한 번 직렬화된 메시지를 다시 한번 직렬화하는, 이중 직렬화 현상이 발생하게 됩니다.

따라서 아래와 같이 재시도 토픽에서 사용할 Kafka Template 을 별도로 한 벌 더 구성하여 retryKafkaTemplate 이라고 명명하고 Bean 으로 등록해 두었습니다.

@Configuration @PropertySource("classpath:common-${spring.profiles.active}.properties") public class KafkaConfig { @Value("${kafka.servers}") private String servers; @Bean public KafkaTemplate<String, Object> kafkaTemplate() { Map<String, Object> configs = new HashMap<>(); configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers); configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(configs)); } @Bean public KafkaTemplate<String, String> retryKafkaTemplate() { Map<String, Object> configs = new HashMap<>(); configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers); configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(configs)); } }

또한 컨슘 실패 시, 알림을 해줘야 하는 정책도 새롭게 세워봤습니다.

처음 메시지가 수신된 메인 토픽에서는 실패 시 개발자에게 알림이 가야 한다.
이후 재시도 토픽에서는 실패하더라도 알림이 가선 안된다.
재시도 토픽에서는 반대로, 메시지 처리에 성공했을 경우 이슈가 해소되었음을 알려야 한다.
혹은 재시도 토픽에서 계속 실패하여 144회 모두 실패할 경우, 데드레터 토픽으로 메시지를 발행하며 이때 최종 실패 알림이 가야 한다.

즉, 현재 컨슘하고 있는 토픽이 메인 토픽인지, 재시도 토픽인지를 판단할 수 있어야 합니다. 이를 위해 메인 토픽 이름과 현재 토픽 이름을 각각 얻어 TopicNameSet 이라는 객체로 생성하여 활용했습니다.

public record TopicNameSet( String mainTopicName, String currentTopicName ) { public static TopicNameSet of(String mainTopicName, String currentTopicName) { return new TopicNameSet(mainTopicName, currentTopicName); } public boolean isMainTopic() { return Objects.equals(this.mainTopicName, this.currentTopicName); } public boolean isRetryTopic() { return !this.isMainTopic(); } }

리스너 코드는 이러한 내용들을 반영하여 최종적으로 아래와 같이 완성되었습니다.

@Component public class SomethingConsumer { private final SomethingConvertUseCase somethingConvertUseCase; private final AlarmUtils alarmUtils; private static final String ATTEMPTS_145_STRING = "145"; private static final long TEN_MINUTES_MILLISECONDS = 600000L; @Value("${spring.kafka.topics.main-topic-name}") private String mainTopicName; public SomethingConsumer( SomethingConvertUseCase somethingConvertUseCase, AlarmUtils alarmUtils ) { this.somethingConvertUseCase = somethingConvertUseCase; this.alarmUtils = alarmUtils; } @RetryableTopic( attempts = ATTEMPTS_145_STRING, backoff = @Backoff(delay = TEN_MINUTES_MILLISECONDS), sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC, kafkaTemplate = "retryKafkaTemplate" ) @KafkaListener(topics = "${spring.kafka.topics.main-topic-name}", containerFactory = "containerFactory") public void onMessage( SomethingRequestDTO somethingRequestDTO, Acknowledgment acknowledgment, @Header(KafkaHeaders.RECEIVED_TOPIC) String currentTopicName ){ TopicNameSet topicNameSet = TopicNameSet.of(mainTopicName, currentTopicName); somethingConvertUseCase.convert(somethingRequestDTO.toDomain(), topicNameSet); acknowledgment.acknowledge(); } @DltHandler public void handleDeadLetter( SomethingRequestDTO somethingRequestDTO, @Header(KafkaHeaders.RECEIVED_TOPIC) String dltTopicName ) { try { alarmUtils.alarm( "[외부 채널 입고 정보 -> 재처리 최종 실패. 데드레터 알림]", "id: " + somethingRequestDTO.id() + "\nDeadLetter Topic Name: " + dltTopicName ); } catch (Exception e) { log.error("외부 채널 입고 정보 데드레터 알람 발송 중 에러가 발생했습니다.", e); } } }

이제 somethingConvertUseCase.convert 메서드 내부에서 예외가 발생하면, RetryableTopic 기능에 의해 자동으로 메인 토픽의 메시지는 커밋되고, 재시도 토픽으로 넘어가 처리되기 시작할 것입니다.

이와 같은 메커니즘들을 통해 저희는 비로소 자유를 만끽할 수 있게 되었습니다.

이렇게 최종적으로 완성된, 아웃박스 패턴과 재시도 토픽이 적용된 입고 데이터 인터페이스 전체 아키텍처를 확인해 보겠습니다.

우리가 흔히 소프트웨어 개발에서 하는 말 중 '바퀴를 다시 발명하지 마라(Don't reinvent the wheel)' 는 격언이 있습니다. 이미 검증된 도구가 있다면 그것을 활용해 본질적인 문제에 집중하라는 뜻이죠.

사실 아키텍처를 설계하며 팀 내부적으로도 많은 고민과 의견이 있었습니다. 처음에는 스케줄링 배치(Batch) 잡을 직접 작성해서, 인터페이스 테이블에 적재된 데이터 중 입고 예정 정보 생성이 완료되지 않은 건들을 처리할까도 생각했습니다. 누락 건을 가장 확실하게 찾아낼 수 있고, 로직이 직관적이라 이해하기 쉽다는 강력한 장점이 있는, 가장 고전적이면서도 믿음직한 방법이니까요.

하지만 이럴 경우, 수신부의 로직이 발행부의 저장소 영역인 인터페이스 테이블에 직접 의존하게 되는 구조가 되어버립니다. 저희는 발행부와 수신부의 책임을 명확히 분리하고 싶었습니다. 발행부는 아웃박스 패턴을 통해 메시지를 안전하게 내보내는 데 집중하고, 수신부는 수신된 메시지를 처리하며 자신의 재시도 상태를 스스로 관리하는 것이 더 견고한 아키텍처라고 판단했습니다.

결국 직접 바퀴를 깎는 대신, 검증된 라이브러리와 프레임워크의 강력한 기능을 사용하는 방안을 채택했습니다. 덕분에 에너지를 3PL 사업의 본질적인 비즈니스 로직을 정교화하는 데 투자할 수 있었죠. 저희의 이 여정이 비슷한 고민을 하는 분들께 조금이나마 도움이 되길 바랍니다.

이상입니다. 긴 글 읽어주셔서 감사합니다. 다른 의견이나 조언, 문의 등 모든 코멘트는 항상 환영합니다.

그럼, 또 뵙겠습니다!

Read Entire Article