Kang Log
Go back

3. 상품 생성 이벤트를 받아 Stock에 초기 재고 생성

Updated:

3. 상품 생성 이벤트를 받아 Stock에 초기 재고 생성

상품 서비스가 발행한 ProductCreated 이벤트를 재고 서비스가 받아 DB와 Redis에 초기 재고를 생성한다. 재고 DB에 처음 등록한 경우에는 StockInitialized 이벤트도 Outbox에 저장한다.

전체 처리 흐름

Product의 OutboxRelay
    ↓
Kafka: product.events
    ↓
Stock의 ProductEventListener
    ↓
메시지 헤더와 JSON 본문을 이벤트 객체로 변환
    ↓
StockService.initializeStock()
    ├─ 기존 재고 있음 → 기존 재고 반환
    └─ 기존 재고 없음
         → 초기 재고 0 저장
         → StockInitialized Outbox 저장
    ↓
DB 트랜잭션 커밋
    ├─ Listener → Redis 초기화
    └─ 별도의 OutboxRelay → stock.events 발행

Redis 초기화와 Relay의 발행은 DB 커밋 이후 독립적으로 진행된다. 따라서 StockInitialized가 Redis 초기화보다 먼저 발행될 수도 있다.

1. Kafka 이벤트 수신 연결

Stock에 ProductEventListener를 작성했다.

@Slf4j
@Component
@RequiredArgsConstructor
public class ProductEventListener {

    private final JsonMapper jsonMapper;
    private final StockService stockService;
    private final StockRedisRepository stockRedisRepository;

    @KafkaListener(
            topics = "product.events",
            groupId = "stock-service"
    )
    public void onProductCreated(ConsumerRecord<String, String> record) {
        Message<ProductCreated> message =
                Message.from(record, ProductCreated.class, jsonMapper);

        ProductCreated event = message.payload();

        Stock stock =
                stockService.initializeStock(event.productId());

        stockRedisRepository.initializeIfAbsent(
                stock.getProductId(),
                stock.getQuantity()
        );

        log.info(
                "상품 재고 초기화 완료: messageId={}, productId={}",
                message.messageId(),
                event.productId()
        );
    }
}

Listener는 다음 역할을 순서대로 수행한다.

  1. Kafka 메시지를 받는다.
  2. 메시지를 이벤트 객체로 변환한다.
  3. Service를 호출하여 DB 재고를 준비한다.
  4. 반환받은 재고로 Redis를 초기화한다.
코드역할
@ComponentListener를 Spring Bean으로 등록
@KafkaListenerKafka 메시지 수신과 메서드 실행 연결
topics구독할 토픽
groupId메시지를 소비할 Consumer Group
ConsumerRecord키·본문·헤더·토픽·파티션·오프셋 등을 담는 객체

Stock과 Order가 서로 다른 그룹을 사용하면 같은 ProductCreated를 각각 받는다. 같은 그룹의 서버들은 파티션을 나누어 처리한다.

2. Kafka 메시지를 이벤트 객체로 변환

Consumer 설정에서는 키와 본문을 문자열로 받도록 지정했다.

spring:
  kafka:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false

Kafka에서 전달받은 바이트를 StringDeserializer가 Java 문자열로 변환한다. 본문 문자열은 아직 JSON이므로, 이후 Message.from()에서 이벤트 객체로 변환한다.

Kafka의 바이트 데이터
    ↓ StringDeserializer
JSON 문자열
    ↓ Message.from() + JsonMapper
ProductCreated 객체
Message<ProductCreated> message =
        Message.from(record, ProductCreated.class, jsonMapper);

ProductCreated event = message.payload();

공통 Message.from()은 헤더와 본문을 나눠 읽는다.

헤더 → messageId, messageType, occurredAt
본문 → ProductCreated

ProductCreated.class는 JSON을 어떤 타입으로 변환할지 지정한다. payload()는 변환된 상품 정보를 반환한다.

auto-offset-reset: earliest는 해당 그룹에 유효한 커밋 오프셋이 없을 때 보관 중인 가장 이른 위치부터 읽도록 한다. 매번 재시작할 때 처음부터 읽는다는 뜻은 아니다.

3. 상품별 재고 조회와 초기 생성

Repository에 상품 ID로 재고를 조회하는 메서드를 추가했다.

public interface StockRepository extends JpaRepository<Stock, Long> {

    Optional<Stock> findByProductId(Long productId);
}

조회 결과가 없을 수 있으므로 Optional<Stock>을 반환한다.

Service는 기존 재고가 있으면 그대로 반환하고, 없을 때만 새로 생성한다.

@Transactional
public Stock initializeStock(Long productId) {
    Optional<Stock> existingStock =
            repository.findByProductId(productId);

    if (existingStock.isPresent()) {
        return existingStock.get();
    }

    Stock stock = repository.save(Stock.createFor(productId));

    outboxWriter.write(
            "stock.events",
            String.valueOf(stock.getProductId()),
            new StockInitialized(
                    stock.getProductId(),
                    stock.getQuantity()
            )
    );

    return stock;
}

Stock.createFor(productId)는 해당 상품의 초기 재고 수량을 0으로 만든다.

상품 ID로 조회
    ├─ 있음 → 기존 객체 반환
    └─ 없음 → 수량 0으로 저장

초기 수량을 상품 생성 요청에서 받지 않고, 이후 재고 증가 API를 통해 추가하도록 구성했다.

4. 재고 등록 이벤트를 Outbox에 저장

재고 DB에 처음 등록했을 때는 StockInitialized를 함께 저장한다.

public record StockInitialized(
        Long productId,
        int quantity
) {
}

ProductCreated와는 발생 주체와 의미가 다르다.

이벤트의미
ProductCreatedProduct에서 상품이 생성되었다
StockInitializedStock에서 상품의 재고 정보가 DB에 등록되었다

재고 저장과 Outbox 저장은 같은 트랜잭션에서 실행한다.

재고 저장 + StockInitialized Outbox 저장
    ├─ 성공 → 함께 커밋
    └─ 실패 → 함께 롤백

기존 재고가 있으면 Outbox 저장 이전에 반환하므로, 같은 상품의 등록 이벤트를 추가 생성하지 않는다.

Stock에도 공통 OutboxConfig를 연결했다. 이후 기존 Relay가 stock.events로 이벤트를 발행한다.

StockInitialized는 DB 재고 등록 사실을 의미한다. Redis 초기화 완료까지 보장하는 이벤트는 아니다.

5. DB 커밋 후 Redis 초기화

Listener는 DB 처리가 끝난 뒤 Redis를 호출한다.

Stock stock =
        stockService.initializeStock(event.productId());

stockRedisRepository.initializeIfAbsent(
        stock.getProductId(),
        stock.getQuantity()
);

현재 Listener에는 바깥 DB 트랜잭션이 없으므로, Service 호출이 성공해서 돌아왔을 때는 재고·Outbox의 커밋까지 끝난 상태다.

StockService 호출
    ↓
재고 + Outbox 저장·커밋
    ↓
Listener로 반환
    ↓
Redis 초기화

Redis 처리를 분리하여 Redis 실패가 이미 저장된 재고·Outbox를 취소하지 않도록 했다. 현재 JPA의 @Transactional이 Redis까지 하나의 트랜잭션으로 묶어 주는 것은 아니다.

6. setIfAbsent()로 기존 Redis 수량 유지

Redis 접근은 StockRedisRepository가 담당한다.

@Repository
@RequiredArgsConstructor
public class StockRedisRepository {

    private final StringRedisTemplate redisTemplate;

    public void initializeIfAbsent(Long productId, int quantity) {
        String key = "stock:available:" + productId;

        redisTemplate.opsForValue()
                .setIfAbsent(key, String.valueOf(quantity));
    }
}

상품별 Redis 키는 다음 형식으로 구성했다.

stock:available:{productId}
코드역할
StringRedisTemplate문자열 키·값으로 Redis에 접근
opsForValue()Redis String 타입의 명령 사용
setIfAbsent()키가 없을 때만 값 저장
String.valueOf(quantity)수량을 문자열로 변환

일반 set()으로 매번 초기화하면, 이벤트를 다시 받았을 때 이미 증가한 재고를 덮어쓸 수 있다.

최초 이벤트 → Redis 0
재고 증가 → Redis 100
같은 이벤트 재수신 → set(0)을 하면 기존 수량 유실

setIfAbsent()를 사용하면 기존 키의 값을 유지한다.

키 없음 → 전달받은 수량으로 초기화
키 있음 → 기존 값 유지

7. 이벤트 재수신과 Redis 실패 처리

동일 이벤트가 다시 들어오면 다음과 같이 처리된다.

기존 DB 재고 조회
    ↓
새 재고·등록 이벤트 생성 없이 반환
    ↓
Redis 초기화 시도
    ├─ 키 있음 → 기존 값 유지
    └─ 키 없음 → 반환받은 DB 수량으로 초기화

DB 저장 후 Redis만 실패한 경우에도 같은 흐름을 활용할 수 있다.

첫 처리
    재고 + Outbox 커밋 성공
    Redis 실패 → Listener 밖으로 예외 전달

재처리
    기존 재고 반환
    새 Outbox는 저장하지 않음
    Redis 초기화 다시 시도

다만 실제 재처리 여부는 Kafka 소비자의 재시도·복구 정책에 따른다. 이 구현 자체가 무제한 복구를 보장하는 것은 아니다.

상품 ID의 DB 유니크 제약도 중복 행 생성을 막는다. 동시에 생성하려는 처리가 충돌하여 한쪽이 롤백되더라도, 먼저 커밋한 재고와 Outbox는 유지된다.

현재는 조회 후 생성하는 단순한 방식으로 진행했다. 키 유실 복구 중 동시 재고 변경까지 안전하게 조정하는 대사 기능은 이후 학습 과제로 남겼다.

검증한 내용

이 단계에서 재고 모듈 테스트 16개가 통과했다. H2와 별도 실제 Redis를 사용했으며, Listener에 메시지를 직접 전달하여 검증했다. 이후 실제 Kafka 연동에서도 초기 재고 저장과 등록 이벤트 수신을 확인했다.

다음 단계에서는 Order도 같은 ProductCreated를 소비하여 상품 스냅샷을 저장하고, 상품 생성의 전체 이벤트 흐름을 완성한다.

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
4. 상품 생성 이벤트를 받아 Order에 상품 스냅샷 저장
Next Post
2. 상품 생성 이벤트를 Outbox에 저장하고 Kafka로 발행