Kang Log
Go back

2. 상품 생성 이벤트를 Outbox에 저장하고 Kafka로 발행

Updated:

2. 상품 생성 이벤트를 Outbox에 저장하고 Kafka로 발행

상품 생성 API에 이벤트 발행을 연결했다. 상품을 저장하면서 ProductCreated 이벤트도 같은 DB 트랜잭션으로 저장하고, Kafka 전송은 기존 OutboxRelay가 별도로 처리한다.

전체 처리 흐름

POST /api/products
    ↓
ProductService.create()
    ├─ 상품 저장
    └─ ProductCreated를 Outbox에 저장
    ↓
DB 트랜잭션 커밋
    ↓
201 Created 응답

별도로 실행되는 OutboxRelay
    ↓
PENDING 이벤트 조회
    ↓
Kafka: product.events로 발행
    ↓
발행 성공 → Outbox를 PUBLISHED로 변경
    ↓
Notification에서 이벤트 수신

API는 Kafka 발행이 끝날 때까지 기다리지 않는다. 상품과 Outbox가 DB에 커밋되면 응답할 수 있고, 실제 발행은 Relay가 담당한다. 따라서 발행과 HTTP 응답의 도착 순서는 고정되지 않는다.

1. 상품 저장 직후 Kafka에 직접 보내지 않은 이유

DB 저장과 Kafka 발행은 서로 다른 시스템의 작업이다. 단순히 순서대로 호출해도 하나의 트랜잭션으로 묶이지 않는다.

상품 DB 저장·커밋 성공
    ↓
Kafka 전송 실패
    ↓
상품은 있지만 다른 서비스는 생성 사실을 모름

반대로 DB 트랜잭션 안에서 Kafka에 먼저 보내면, 이후 DB 롤백에도 이미 보낸 이벤트는 남을 수 있다.

이를 피하기 위해 상품과 발행할 이벤트를 먼저 같은 DB에 저장한다.

상품 + Outbox
    ├─ 함께 커밋
    └─ 함께 롤백

Outbox는 외부로 발행할 이벤트를 기록해 두는 DB 테이블이다. Kafka 전송이 실패해도 발행할 내용이 DB에 남으므로 다시 시도할 수 있다.

2. ProductCreated 이벤트 정의

상품 생성 사실과 다른 서비스가 사용할 정보를 이벤트에 담았다.

public record ProductCreated(
        Long productId,
        String name,
        long price
) {
}
필드의미
productId생성된 상품 ID
name상품명
price원 단위 가격

Stock은 상품 ID로 초기 재고를 만들고, Order는 상품 ID·이름·가격으로 상품 스냅샷을 만들 수 있다.

엔티티의 가격은 BigDecimal로 관리하고, 이벤트에는 원 단위 정수인 long으로 변환해서 담는다.

3. ProductService에 Outbox 저장 연결

기존 상품 저장 메서드에 OutboxWriter 호출을 추가했다.

@Service
@RequiredArgsConstructor
public class ProductService {

    private final ProductRepository repository;
    private final OutboxWriter outboxWriter;

    @Transactional
    public Product create(String name, BigDecimal price) {
        Product product = Product.register(name, price);

        try {
            product = repository.saveAndFlush(product);
        } catch (DataIntegrityViolationException e) {
            throw new BusinessException(
                    ErrorCode.PRODUCT_NAME_DUPLICATED
            );
        }

        outboxWriter.write(
                "product.events",
                String.valueOf(product.getId()),
                new ProductCreated(
                        product.getId(),
                        product.getName(),
                        product.getPrice().longValueExact()
                )
        );

        return product;
    }
}

상품을 먼저 저장하여 생성된 ID를 확보한 뒤, 그 ID를 이벤트에 사용한다.

outboxWriter.write(
        "product.events",                    // 발행할 토픽
        String.valueOf(product.getId()),     // 메시지 키
        new ProductCreated(...)              // 이벤트 본문
);

이 호출은 Kafka로 즉시 전송하는 것이 아니라 Outbox에 저장하는 작업이다.

4. 공통 Message와 OutboxWriter 활용

이벤트마다 직접 JSON 변환과 메타데이터 생성을 반복하지 않도록 기존 공통 기능을 활용했다.

Message는 이벤트 본문과 함께 다음 정보를 다룬다.

항목역할
messageId메시지 식별자
messageType이벤트 종류
occurredAt이벤트 발생 시각
payload실제 이벤트 데이터

ProductCreated가 상품 정보를 담는 객체라면, Message<ProductCreated>는 상품 정보에 메시지 관리 정보를 붙인 형태다.

OutboxWriter의 처리 흐름은 다음과 같다.

이벤트 객체 전달
    ↓
Message.of(payload)
    ↓
메시지 ID·종류·발생 시각 생성
    ↓
payload를 JSON 문자열로 변환
    ↓
OutboxMessage 생성
    ↓
DB 저장

이때 사용하는 코드가 다음과 같다.

jsonMapper.writeValueAsString(payload)

Java 객체를 JSON 문자열로 변환하는 직렬화 작업이다.

new ProductCreated(1L, "키보드", 30000L)
{
  "productId": 1,
  "name": "키보드",
  "price": 30000
}

이 JSON과 토픽·키·메시지 정보를 Outbox에 기록한다.

5. 상품과 Outbox의 트랜잭션 연결

ProductService.create()에는 @Transactional이 있고, OutboxWriter.write()는 기존 트랜잭션 참여를 요구한다.

@Transactional(propagation = Propagation.MANDATORY)

MANDATORY는 이미 시작된 트랜잭션이 있어야 실행할 수 있다는 의미다. 트랜잭션 없이 호출하면 예외가 발생한다.

현재 호출에서는 상품 서비스의 트랜잭션에 참여한다.

ProductService 트랜잭션 시작
    ↓
상품 저장
    ↓
OutboxWriter가 같은 트랜잭션에서 이벤트 저장
    ↓
함께 커밋

따라서 Outbox 저장이 실패하면 상품 저장도 롤백된다.

앞에서 실행한 saveAndFlush()가 성공했더라도 아직 커밋한 것은 아니므로, 상품만 따로 남지 않는다.

6. 기존 OutboxRelay로 Kafka 발행

Outbox에 저장된 이벤트는 처음에 PENDING 상태다. 기존 Relay가 주기적으로 미발행 이벤트를 조회해 Kafka로 전송한다.

PENDING 조회
    ↓
토픽·키·본문·헤더를 사용해 Kafka 메시지 생성
    ↓
Kafka 전송 결과 확인
    ├─ 성공 → PUBLISHED로 변경
    └─ 실패 → PENDING 유지

Kafka 메시지는 다음과 같이 구성된다.

Topic: product.events
Key: 상품 ID
Headers: messageId, messageType, occurredAt
Value: ProductCreated의 JSON

발행 실패 시 이벤트를 미발행 상태로 남겨 다음 실행에서 다시 시도할 수 있게 한다.

단, Outbox가 중복 발행까지 완전히 막는 것은 아니다.

Kafka 전송 성공
    ↓
PUBLISHED 저장 전에 서버 종료
    ↓
재시작 후 같은 이벤트 재발행 가능

이 때문에 다음 소비 단계에서는 같은 이벤트를 다시 받아도 기존 데이터를 잘못 변경하지 않도록 처리해야 한다.

7. OutboxConfig와 ShedLock 연결

상품 애플리케이션에서 공통 Outbox 설정을 사용하도록 연결했다.

@Import({
        GlobalExceptionHandler.class,
        OutboxConfig.class
})

OutboxConfig는 Outbox 관련 Bean과 스케줄러, ShedLock 설정을 연결한다.

구성역할
컴포넌트 스캔OutboxWriter, OutboxRelay 등 등록
@EnableScheduling주기적으로 Relay 실행
@EnableSchedulerLock스케줄 작업에 분산 락 적용
LockProviderDB를 이용한 락 획득·해제 처리

ShedLock은 같은 서비스의 서버가 여러 대 실행될 때, 같은 Relay 작업이 동시에 실행되는 것을 줄이기 위한 장치다.

같은 서비스의 서버 A·B
    ↓
공유 DB의 shedlock 테이블에서 같은 이름의 락 획득 시도
    ├─ A 획득 성공 → Relay 실행
    └─ B 획득 실패 → 이번 실행 건너뜀

서버 이름으로 순서를 정하는 것이 아니라, DB에서 락 획득에 성공한 서버가 실행한다. usingDbTime()은 서버별 시계 대신 DB 시간을 사용하도록 한다.

lockAtMostFor = "PT1M"은 서버가 비정상 종료해도 잠금이 영원히 남지 않도록 두는 최대 유지 시간이다. 1분이 지나면 작업을 강제로 종료하는 설정은 아니다. 작업이 그보다 오래 실행되면 다른 서버가 락을 얻을 수 있으므로, 실행 시간에 맞는 설정은 이후 조정할 요소로 남겼다.

검증한 내용

상품 모듈은 Outbox 연결 이후 47개 테스트가 통과했다. 이후 실제 서비스 연동에서도 Notification의 ProductCreated 수신과 Outbox의 PUBLISHED 상태를 확인했다.

여기까지 상품 생성 사실을 DB에 안전하게 기록하고, 기존 Relay를 통해 Kafka로 전달하는 흐름을 연결했다. 다음 단계에서는 Stock이 이 이벤트를 소비하여 초기 재고를 만든다.

Share this post on:
Share this post via WhatsApp Share this post on Facebook Share this post on X Share this post via Telegram Share this post on Pinterest Share this post via email
Previous Post
3. 상품 생성 이벤트를 받아 Stock에 초기 재고 생성
Next Post
1. 상품 생성 API 구현