Spring으로 MSA 구현하기 3편 [Kafka, Gateway]

Spring으로 MSA 구현하기 1편 [gRPC]

Spring으로 MSA 구현하기 2편 [Outbox]

프로젝트 코드

GitHub – dalmengs/kotlin-spring-msa-wallet

GitHub – dalmengs/kotlin-spring-msa-order

GitHub – dalmengs/convention_proto

Kafka 연동

이전 글까지의 구현을 정리해보면, 결제 요청이 들어왔을 때 다음 단계까지는 이미 완료된 상태다.

재고 감소
결제 성공 이벤트 생성
해당 이벤트를 Outbox 테이블에 안전하게 저장
Scheduler를 통해 주기적으로 Outbox 이벤트를 조회
 이제는 Publisher에서 저장된 이벤트를 실제로 Kafka에 발행하고, 각 서비스에서 이를 비동기적으로 소비하는 것을 구현해야 한다.

그 다음으로는 각 서비스에 Kafka Listener를 붙여, 결제 완료 이벤트를 비동기적으로 수신한다. 

Order 서비스는 이 이벤트를 받아 주문 상태를 완료로 변경하고, 그 외의 서비스들도 각자의 책임에 맞는 후처리를 수행하도록 할 예정이다.

Kafka 환경 세팅

Kafka는 Docker Compose로 가볍게 띄워서 테스트 하였다.

YAML

version: "3.8"

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

version: "3.8"

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.5.0
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.5.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

아래 커맨드로 Kafka Topic을 만들 수 있다.

Zsh

docker exec -it kafka kafka-topics \
  --bootstrap-server localhost:9092 \
  --create \
  --topic wallet.debited \
  --partitions 1 \
  --replication-factor 1

docker exec -it kafka kafka-topics \
  --bootstrap-server localhost:9092 \
  --create \
  --topic wallet.debited \
  --partitions 1 \
  --replication-factor 1

Topic을 만든 후 아래 커맨드로 잘 만들어졌는지 확인할 수 있다.

Zsh

docker exec -it kafka kafka-topics \
  --bootstrap-server localhost:9092 \
  --list

# [Standard Output]
# wallet.debited

docker exec -it kafka kafka-topics \
  --bootstrap-server localhost:9092 \
  --list

# [Standard Output]
# wallet.debited

Spring Kafka 세팅

Spring Datasource 설정하듯이 application.yml에 설정해주면 된다.

YAML

spring:
  kafka:
    bootstrap-servers: localhost:9092

    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      auto-offset-reset: earliest

spring:
  kafka:
    bootstrap-servers: localhost:9092

    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer:
      auto-offset-reset: earliest

Kafka 이벤트에 넣을 데이터와 Topic을 정의해준다.

Kotlin

data class WalletDebitedEvent(
    val userId: String,
    val orderId: String
)

data class WalletDebitedEvent(
    val userId: String,
    val orderId: String
)

Kotlin

object WalletTopics {
    const val PAYMENT_WALLET_DEBITED = "wallet.debited"
}

object WalletTopics {
    const val PAYMENT_WALLET_DEBITED = "wallet.debited"
}

Event Publish

KafkaTemplate을 주입받아 Polling Scheduler 에서 Kafka에 이벤트를 보낼 수 있다.

Kotlin

@Transactional
fun publishEvent(outboxId: String) {
    val outbox = outboxRepository.findPendingByIdWithLock(outboxId)
        ?: return

    try {
        kafkaTemplate.send(
            outbox.eventType,
            outbox.aggregateId,
            outbox.payload
        )

        outbox.makePublished()
    } catch (e: Exception) {
        outbox.makeFailed()
    }
}

@Transactional
fun publishEvent(outboxId: String) {
    val outbox = outboxRepository.findPendingByIdWithLock(outboxId)
        ?: return

    try {
        kafkaTemplate.send(
            outbox.eventType,
            outbox.aggregateId,
            outbox.payload
        )

        outbox.makePublished()
    } catch (e: Exception) {
        outbox.makeFailed()
    }
}

Event Listening

Kafka에서는 @KafkaListener 어노테이션을 통해 이벤트를 소비하는 핸들러를 간단하게 등록할 수 있다.

Outbox Publisher를 통해 Kafka로 발행된 이벤트는 각 서비스에서 Listener를 통해 비동기적으로 수신된다. 
메시지는 문자열 형태로 전달되므로, ObjectMapper를 이용해 역직렬화하여 미리 정의해둔 WalletDebitedEvent와 같은 도메인 이벤트 객체로 변환한다.

아래 Listener는 Order Service에서 사용되는 Consumer 예시다. wallet.debited 이벤트를 받아 해당 주문을 최종 완료 처리하는 역할을 한다.

Kotlin

@Component
class KafkaWalletConsumer(
    private val objectMapper: ObjectMapper,
    private val orderService: OrderService,
) {

    @KafkaListener(
        topics = ["wallet.debited"],
        groupId = "order-service"
    )
    fun consume(message: String) {
        try {
            val event = objectMapper.readValue(
                message,
                WalletDebitedEvent::class.java
            )

            orderService.markOrderAsPaid(event.userId, event.orderId)

        } catch (e: Exception) {
            throw e
        }
    }
}

@Component
class KafkaWalletConsumer(
    private val objectMapper: ObjectMapper,
    private val orderService: OrderService,
) {

    @KafkaListener(
        topics = ["wallet.debited"],
        groupId = "order-service"
    )
    fun consume(message: String) {
        try {
            val event = objectMapper.readValue(
                message,
                WalletDebitedEvent::class.java
            )

            orderService.markOrderAsPaid(event.userId, event.orderId)

        } catch (e: Exception) {
            throw e
        }
    }
}

전체 플로우 정리

이제 전체 흐름을 다시 정리해보면 다음과 같다.

1. `POST /api/order` -> 주문 생성 (결제 대기 상태)

2. `POST /api/order/{orderId}` -> 결제 요청

Wallet Service
- 재고 차감
- Outbox에 wallet.debited 이벤트 저장

3. Outbox Event Publisher (Scheduler)

Kafka로 이벤트 발행

4. Kafka Listener (Order Service)

주문 상태를 PAID로 변경

동기 트랜잭션은 결제와 재고 차감까지만, 그 이후의 모든 처리는 비동기 이벤트 기반을 구현하여 MSA다운 구조가 완성되었다.

Gateway

API Gateway로는 Spring Cloud Gateway를 사용했다. 
Gateway 자체의 구현 방식이나 인증 필터 구조는 이전에 정리해둔 프로젝트의 구성을 그대로 재사용하였다.

[MSA] Spring Cloud Gateway

Gateway에서는 외부에서 들어오는 요청을 각 서비스로 적절하게 포워딩하도록 라우트를 정의한다.
아래 설정은 Wallet 서비스와 Order 서비스로 요청을 분기하는 예시다.

YAML

spring:
  cloud:
    gateway:
      server:
        webflux:
          routes:
            - id: wallet-service
              uri: http://localhost:8081
              predicates:
                - Path=/api/wallet/**
            - id: order-service
              uri: http://localhost:8082
              predicates:
                - Path=/api/order/**
server:
  port: 8080

spring:
  cloud:
    gateway:
      server:
        webflux:
          routes:
            - id: wallet-service
              uri: http://localhost:8081
              predicates:
                - Path=/api/wallet/**
            - id: order-service
              uri: http://localhost:8082
              predicates:
                - Path=/api/order/**
server:
  port: 8080

이제 각 서비스는 더 이상 외부 트래픽을 직접 받지 않고, 모든 요청은 반드시 8080 Port Gateway를 거쳐서만 진입하도록 구성했다.

테스트

Gateway 서버인 8080 포트로 주문을 생성하는 API를 날린다.

http://127.0.0.1:8080/api/order
게이트웨이 서버는 Bearer Token 을 Decode 해서 인증 과정을 거친 후 헤더에 X-User-Id를 담아 요청한다.
status = CREATED로 주문이 생성되고, 결제 대기 중인 상태이다.

다음으로는 /api/order/{odrder_id}로 결제 API를 날린다.

재고 감소 성공 후 paymentResult = SUCCESS로 반환된다

다른 케이스로, 잔액이 부족할 때 경우를 테스트해보았다.
주문을 생성할 때 amount = 999999로 넣어서 엄청난 가격의 주문을 생성했다.

실제 서비스에서는 절대 amount 같은 값을 Request Data로 받도록 하면 안 된다..!

주문 생성은 잔액과 상관 없기 때문에 정상적으로 생성되는 것을 알 수 있다.

하지만 결제 API를 날리니 INSUFFICIENT_BALANCE로 실패한 것을 알 수 있다.

이때는 FAILED로 바꾸지 않고, 아무 이벤트도 날리지 않아 계속 CREATED 상태에 머무르도록 한다.
잔액 충전 후 결제 할 수도 있으니깐…

Kafka 연동부터 Gateway 서버까지 모두 연결하면서, 이번 프로젝트에서는 전체 결제 플로우를 MSA 구조로 끝까지 처리할 수 있는 기반을 완성했다.

실제 서비스 환경에서는 이보다 훨씬 더 많은 마이크로서비스와 이벤트들이 얽혀 있지만, 이 정도 규모의 프로젝트만으로도 gRPC 기반 내부 통신, Outbox 패턴을 활용한 안정적인 이벤트 관리, Kafka를 이용한 비동기 처리 흐름을 모두 경험해볼 수 있다.

실무에서 바로 활용할 수 있는 설계 포인트와 고민들을 직접 구현해볼 수 있다는 점에서 이번 구조는 충분히 의미 있는 연습이라고 생각한다.

다음 글에서는 결제 도메인에서 특히 주의해야 할 트랜잭션 설계 포인트와 함께, 동시성 테스트를 통해 여러 결제 요청이 동시에 들어오는 상황에서도 데이터 정합성이 깨지지 않고 안정적으로 처리되는 구조를 검증해볼 예정이다.