CQRS + Event Sourcing 마이크로서비스 아키텍처: Kafka 기반 구현과 Saga 패턴 실전 가이드

OrderCreated { orderId: "ORD-001", customerId: "CUST-42", items: [...] }
  → OrderItemAdded { orderId: "ORD-001", productId: "PROD-7", qty: 2 }
    → PaymentReceived { orderId: "ORD-001", amount: 59000, method: "CARD" }
      → OrderShipped { orderId: "ORD-001", trackingNumber: "KR1234567890" }

[Snapshot at version 50: { status: "PAID", total: 59000, ... }]
  → [Event 51] OrderItemRemoved { ... }
  → [Event 52] RefundInitiated { ... }
  → ... (현재 버전까지)

# kafka-topics.yml - 토픽 설계 예시
topics:
  # 주문 Aggregate 이벤트 토픽
  - name: order-events
    partitions: 12
    replication-factor: 3
    config:
      retention.ms: -1 # 무기한 보존 (이벤트 스토어)
      cleanup.policy: delete # 이벤트 로그는 삭제 정책
      min.insync.replicas: 2 # 최소 2개 ISR 유지
      max.message.bytes: 1048576 # 1MB 메시지 크기 제한
      message.timestamp.type: CreateTime

  # 주문 스냅샷 토픽 (Log Compaction)
  - name: order-snapshots
    partitions: 12
    replication-factor: 3
    config:
      cleanup.policy: compact # 키별 최신 메시지만 유지
      min.compaction.lag.ms: 86400000 # 최소 24시간 후 compaction
      delete.retention.ms: 604800000 # 삭제 마커 7일 유지

  # Saga 상태 토픽
  - name: order-saga-events
    partitions: 12
    replication-factor: 3
    config:
      retention.ms: 2592000000 # 30일 보존
      cleanup.policy: delete

// build.gradle.kts
dependencies {
    implementation("org.springframework.boot:spring-boot-starter-web")
    implementation("org.springframework.kafka:spring-kafka")
    implementation("com.fasterxml.jackson.module:jackson-module-kotlin")
    implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310")
}

// application.yml
// spring:
//   kafka:
//     bootstrap-servers: kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092
//     producer:
//       acks: all
//       retries: 3
//       key-serializer: org.apache.kafka.common.serialization.StringSerializer
//       value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
//       properties:
//         enable.idempotence: true
//         max.in.flight.requests.per.connection: 5
//     consumer:
//       group-id: order-projection
//       auto-offset-reset: earliest
//       enable-auto-commit: false
//       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
//       value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

// EventPublisher.kt - Kafka 기반 이벤트 발행
@Component
class KafkaEventPublisher(
    private val kafkaTemplate: KafkaTemplate<String, DomainEvent>,
    private val objectMapper: ObjectMapper
) {
    private val logger = LoggerFactory.getLogger(javaClass)

    /**
     * 도메인 이벤트를 Kafka 토픽에 발행한다.
     * Aggregate ID를 키로 사용하여 파티션 내 순서를 보장한다.
     */
    suspend fun publish(event: DomainEvent): RecordMetadata {
        val topic = resolveTopicName(event)
        val key = event.aggregateId

        val record = ProducerRecord<String, DomainEvent>(topic, key, event).apply {
            headers().add("eventType", event.eventType.toByteArray())
            headers().add("eventVersion", event.version.toString().toByteArray())
            headers().add("correlationId", event.metadata.correlationId.toByteArray())
            headers().add("timestamp", Instant.now().toString().toByteArray())
        }

        return try {
            val result = kafkaTemplate.send(record).get()
            logger.info(
                "이벤트 발행 완료: topic={}, partition={}, offset={}, key={}",
                result.recordMetadata.topic(),
                result.recordMetadata.partition(),
                result.recordMetadata.offset(),
                key
            )
            result.recordMetadata
        } catch (e: Exception) {
            logger.error("이벤트 발행 실패: topic={}, key={}, error={}", topic, key, e.message)
            throw EventPublishException("이벤트 발행에 실패했습니다: ${e.message}", e)
        }
    }

    private fun resolveTopicName(event: DomainEvent): String {
        return when (event) {
            is OrderEvent -> "order-events"
            is PaymentEvent -> "payment-events"
            is InventoryEvent -> "inventory-events"
            else -> throw IllegalArgumentException("알 수 없는 이벤트 타입: ${event::class}")
        }
    }
}

// OrderProjectionConsumer.kt - 읽기 모델 프로젝션
@Component
class OrderProjectionConsumer(
    private val orderReadRepository: OrderReadRepository,
    private val acknowledgment: Acknowledgment?  // 수동 커밋
) {
    private val logger = LoggerFactory.getLogger(javaClass)

    @KafkaListener(
        topics = ["order-events"],
        groupId = "order-read-projection",
        containerFactory = "kafkaListenerContainerFactory"
    )
    fun onOrderEvent(
        record: ConsumerRecord<String, DomainEvent>,
        ack: Acknowledgment
    ) {
        val event = record.value()
        val aggregateId = record.key()

        logger.debug(
            "이벤트 수신: type={}, aggregateId={}, offset={}",
            event.eventType, aggregateId, record.offset()
        )

        try {
            when (event) {
                is OrderCreated -> handleOrderCreated(event)
                is OrderItemAdded -> handleOrderItemAdded(event)
                is PaymentReceived -> handlePaymentReceived(event)
                is OrderShipped -> handleOrderShipped(event)
                is OrderCancelled -> handleOrderCancelled(event)
            }
            // 프로젝션 성공 후 오프셋 수동 커밋
            ack.acknowledge()
        } catch (e: Exception) {
            logger.error(
                "프로젝션 실패: type={}, aggregateId={}, error={}",
                event.eventType, aggregateId, e.message
            )
            // 재시도 로직 또는 DLQ로 전송
            throw e
        }
    }

    private fun handleOrderCreated(event: OrderCreated) {
        val orderView = OrderReadModel(
            orderId = event.aggregateId,
            customerId = event.customerId,
            status = OrderStatus.CREATED,
            totalAmount = BigDecimal.ZERO,
            items = mutableListOf(),
            createdAt = event.timestamp,
            updatedAt = event.timestamp
        )
        orderReadRepository.save(orderView)
    }

    private fun handlePaymentReceived(event: PaymentReceived) {
        orderReadRepository.findById(event.aggregateId)?.let { order ->
            order.status = OrderStatus.PAID
            order.paymentMethod = event.method
            order.paidAmount = event.amount
            order.updatedAt = event.timestamp
            orderReadRepository.save(order)
        }
    }

    // 나머지 핸들러 구현 생략
}

// DomainEvent.kt - 도메인 이벤트 기반 클래스
sealed class DomainEvent(
    open val aggregateId: String,
    open val eventType: String,
    open val version: Int,
    open val timestamp: Instant,
    open val metadata: EventMetadata
)

data class EventMetadata(
    val correlationId: String = UUID.randomUUID().toString(),
    val causationId: String = "",
    val userId: String = ""
)

// OrderEvents.kt - 주문 도메인 이벤트
sealed class OrderEvent : DomainEvent()

data class OrderCreated(
    override val aggregateId: String,
    val customerId: String,
    val items: List<OrderItemData>,
    override val version: Int = 1,
    override val timestamp: Instant = Instant.now(),
    override val metadata: EventMetadata = EventMetadata()
) : OrderEvent() {
    override val eventType: String = "OrderCreated"
}

data class PaymentReceived(
    override val aggregateId: String,
    val amount: BigDecimal,
    val method: String,
    val transactionId: String,
    override val version: Int = 1,
    override val timestamp: Instant = Instant.now(),
    override val metadata: EventMetadata = EventMetadata()
) : OrderEvent() {
    override val eventType: String = "PaymentReceived"
}

data class OrderShipped(
    override val aggregateId: String,
    val trackingNumber: String,
    val carrier: String,
    override val version: Int = 1,
    override val timestamp: Instant = Instant.now(),
    override val metadata: EventMetadata = EventMetadata()
) : OrderEvent() {
    override val eventType: String = "OrderShipped"
}

data class OrderCancelled(
    override val aggregateId: String,
    val reason: String,
    val cancelledBy: String,
    override val version: Int = 1,
    override val timestamp: Instant = Instant.now(),
    override val metadata: EventMetadata = EventMetadata()
) : OrderEvent() {
    override val eventType: String = "OrderCancelled"
}

// OrderAggregate.kt - 이벤트 소싱 기반 Aggregate
class OrderAggregate private constructor() {
    lateinit var orderId: String
        private set
    lateinit var customerId: String
        private set
    var status: OrderStatus = OrderStatus.DRAFT
        private set
    var totalAmount: BigDecimal = BigDecimal.ZERO
        private set
    val items: MutableList<OrderItem> = mutableListOf()
    var version: Long = 0
        private set

    // 미발행 이벤트 목록
    private val uncommittedEvents: MutableList<DomainEvent> = mutableListOf()

    companion object {
        /**
         * 이벤트 스트림에서 Aggregate 상태를 복원한다.
         */
        fun rehydrate(events: List<DomainEvent>): OrderAggregate {
            val aggregate = OrderAggregate()
            events.forEach { event -> aggregate.apply(event) }
            return aggregate
        }
    }

    // ──────────── Command Handlers ────────────

    fun createOrder(command: CreateOrderCommand): OrderCreated {
        require(status == OrderStatus.DRAFT) {
            "이미 생성된 주문입니다: orderId=$orderId"
        }
        val event = OrderCreated(
            aggregateId = command.orderId,
            customerId = command.customerId,
            items = command.items,
            metadata = EventMetadata(
                correlationId = command.correlationId,
                userId = command.userId
            )
        )
        applyAndRecord(event)
        return event
    }

    fun receivePayment(command: ReceivePaymentCommand): PaymentReceived {
        require(status == OrderStatus.CREATED) {
            "결제 가능한 상태가 아닙니다: status=$status"
        }
        require(command.amount >= totalAmount) {
            "결제 금액이 부족합니다: required=$totalAmount, received=${command.amount}"
        }
        val event = PaymentReceived(
            aggregateId = orderId,
            amount = command.amount,
            method = command.method,
            transactionId = command.transactionId
        )
        applyAndRecord(event)
        return event
    }

    fun cancelOrder(command: CancelOrderCommand): OrderCancelled {
        require(status != OrderStatus.SHIPPED && status != OrderStatus.CANCELLED) {
            "취소할 수 없는 상태입니다: status=$status"
        }
        val event = OrderCancelled(
            aggregateId = orderId,
            reason = command.reason,
            cancelledBy = command.cancelledBy
        )
        applyAndRecord(event)
        return event
    }

    // ──────────── Event Handlers (상태 변경) ────────────

    private fun apply(event: DomainEvent) {
        when (event) {
            is OrderCreated -> {
                orderId = event.aggregateId
                customerId = event.customerId
                status = OrderStatus.CREATED
                totalAmount = event.items.sumOf { it.price * it.quantity.toBigDecimal() }
                items.addAll(event.items.map { OrderItem(it.productId, it.quantity, it.price) })
            }
            is PaymentReceived -> {
                status = OrderStatus.PAID
            }
            is OrderShipped -> {
                status = OrderStatus.SHIPPED
            }
            is OrderCancelled -> {
                status = OrderStatus.CANCELLED
            }
            else -> { /* 알 수 없는 이벤트는 무시 */ }
        }
        version++
    }

    private fun applyAndRecord(event: DomainEvent) {
        apply(event)
        uncommittedEvents.add(event)
    }

    fun getUncommittedEvents(): List<DomainEvent> = uncommittedEvents.toList()

    fun markEventsAsCommitted() {
        uncommittedEvents.clear()
    }
}

enum class OrderStatus {
    DRAFT, CREATED, PAID, SHIPPED, DELIVERED, CANCELLED
}

// OrderCommandHandler.java - 커맨드 처리 서비스 (Java 예제)
@Service
@Transactional
public class OrderCommandHandler {

    private final KafkaEventPublisher eventPublisher;
    private final OrderEventStore eventStore;
    private final SnapshotStore snapshotStore;

    private static final int SNAPSHOT_INTERVAL = 50;

    public OrderCommandHandler(
            KafkaEventPublisher eventPublisher,
            OrderEventStore eventStore,
            SnapshotStore snapshotStore) {
        this.eventPublisher = eventPublisher;
        this.eventStore = eventStore;
        this.snapshotStore = snapshotStore;
    }

    public String handleCreateOrder(CreateOrderCommand command) {
        // 1. 새로운 Aggregate 생성
        OrderAggregate aggregate = OrderAggregate.rehydrate(Collections.emptyList());

        // 2. 커맨드 실행 (도메인 불변식 검증 포함)
        OrderCreated event = aggregate.createOrder(command);

        // 3. 이벤트를 Kafka에 발행
        eventPublisher.publish(event);

        // 4. 이벤트 커밋 완료 마킹
        aggregate.markEventsAsCommitted();

        return event.getAggregateId();
    }

    public void handleReceivePayment(ReceivePaymentCommand command) {
        // 1. 이벤트 스토어에서 Aggregate 복원
        OrderAggregate aggregate = loadAggregate(command.getOrderId());

        // 2. 커맨드 실행
        PaymentReceived event = aggregate.receivePayment(command);

        // 3. 이벤트 발행
        eventPublisher.publish(event);

        // 4. 스냅샷 조건 확인
        if (aggregate.getVersion() % SNAPSHOT_INTERVAL == 0) {
            snapshotStore.saveSnapshot(aggregate);
        }

        aggregate.markEventsAsCommitted();
    }

    private OrderAggregate loadAggregate(String aggregateId) {
        // 스냅샷이 있으면 스냅샷부터 로드
        Optional<Snapshot> snapshot = snapshotStore.findLatest(aggregateId);
        long fromVersion = snapshot.map(Snapshot::getVersion).orElse(0L);

        // 스냅샷 이후의 이벤트만 로드
        List<DomainEvent> events = eventStore.loadEvents(aggregateId, fromVersion);

        if (snapshot.isPresent()) {
            OrderAggregate aggregate = snapshot.get().toAggregate();
            events.forEach(aggregate::apply);
            return aggregate;
        }

        return OrderAggregate.rehydrate(events);
    }
}

// OrderSagaOrchestrator.kt - 주문 처리 Saga
@Component
class OrderSagaOrchestrator(
    private val kafkaTemplate: KafkaTemplate<String, SagaCommand>,
    private val sagaStateStore: SagaStateStore
) {
    private val logger = LoggerFactory.getLogger(javaClass)

    /**
     * Saga 시작: 주문 생성 이벤트를 수신하면 Saga를 시작한다.
     */
    @KafkaListener(topics = ["order-events"], groupId = "order-saga-orchestrator")
    fun onOrderCreated(record: ConsumerRecord<String, DomainEvent>, ack: Acknowledgment) {
        val event = record.value()
        if (event !is OrderCreated) return

        val sagaId = UUID.randomUUID().toString()
        val sagaState = SagaState(
            sagaId = sagaId,
            orderId = event.aggregateId,
            currentStep = SagaStep.RESERVE_INVENTORY,
            status = SagaStatus.STARTED,
            startedAt = Instant.now()
        )
        sagaStateStore.save(sagaState)

        // Step 1: 재고 예약 커맨드 발행
        val reserveCommand = ReserveInventoryCommand(
            sagaId = sagaId,
            orderId = event.aggregateId,
            items = event.items
        )
        kafkaTemplate.send("inventory-commands", event.aggregateId, reserveCommand)
        logger.info("Saga 시작: sagaId={}, orderId={}, step=RESERVE_INVENTORY", sagaId, event.aggregateId)
        ack.acknowledge()
    }

    /**
     * Step 2: 재고 예약 성공 시 결제 요청
     */
    @KafkaListener(topics = ["inventory-events"], groupId = "order-saga-orchestrator")
    fun onInventoryReserved(record: ConsumerRecord<String, DomainEvent>, ack: Acknowledgment) {
        val event = record.value()
        if (event !is InventoryReserved) return

        val sagaState = sagaStateStore.findBySagaId(event.sagaId) ?: return
        sagaState.currentStep = SagaStep.PROCESS_PAYMENT
        sagaStateStore.save(sagaState)

        val paymentCommand = ProcessPaymentCommand(
            sagaId = event.sagaId,
            orderId = sagaState.orderId,
            amount = sagaState.totalAmount
        )
        kafkaTemplate.send("payment-commands", sagaState.orderId, paymentCommand)
        logger.info("Saga 진행: sagaId={}, step=PROCESS_PAYMENT", event.sagaId)
        ack.acknowledge()
    }

    /**
     * 보상 트랜잭션: 결제 실패 시 재고 예약 취소
     */
    @KafkaListener(topics = ["payment-events"], groupId = "order-saga-orchestrator")
    fun onPaymentFailed(record: ConsumerRecord<String, DomainEvent>, ack: Acknowledgment) {
        val event = record.value()
        if (event !is PaymentFailed) return

        val sagaState = sagaStateStore.findBySagaId(event.sagaId) ?: return
        sagaState.status = SagaStatus.COMPENSATING
        sagaStateStore.save(sagaState)

        // 보상: 재고 예약 취소
        val cancelReservation = CancelInventoryReservationCommand(
            sagaId = event.sagaId,
            orderId = sagaState.orderId
        )
        kafkaTemplate.send("inventory-commands", sagaState.orderId, cancelReservation)

        // 보상: 주문 취소
        val cancelOrder = CancelOrderCommand(
            orderId = sagaState.orderId,
            reason = "결제 실패: ${event.failureReason}",
            cancelledBy = "SAGA_ORCHESTRATOR"
        )
        kafkaTemplate.send("order-commands", sagaState.orderId, cancelOrder)

        logger.warn("Saga 보상 실행: sagaId={}, reason={}", event.sagaId, event.failureReason)
        ack.acknowledge()
    }
}

# Consumer Lag 모니터링 스크립트
#!/bin/bash

BOOTSTRAP_SERVERS="kafka-broker-1:9092,kafka-broker-2:9092"
CONSUMER_GROUPS=("order-read-projection" "order-saga-orchestrator" "search-indexer")
ALERT_THRESHOLD=10000

for GROUP in "${CONSUMER_GROUPS[@]}"; do
    echo "=== Consumer Group: $GROUP ==="

    # 컨슈머 그룹 상태 조회
    kafka-consumer-groups.sh \
        --bootstrap-server "$BOOTSTRAP_SERVERS" \
        --describe \
        --group "$GROUP" 2>/dev/null

    # 총 Lag 계산
    TOTAL_LAG=$(kafka-consumer-groups.sh \
        --bootstrap-server "$BOOTSTRAP_SERVERS" \
        --describe \
        --group "$GROUP" 2>/dev/null \
        | tail -n +3 \
        | awk '{sum += $6} END {print sum}')

    echo "Total Lag: $TOTAL_LAG"

    # 임계값 초과 시 알림
    if [ "$TOTAL_LAG" -gt "$ALERT_THRESHOLD" ]; then
        echo "[ALERT] Consumer group $GROUP lag ($TOTAL_LAG) exceeds threshold ($ALERT_THRESHOLD)"
        # Slack 웹훅 또는 PagerDuty 호출
        curl -s -X POST "$SLACK_WEBHOOK_URL" \
            -H 'Content-Type: application/json' \
            -d "{\"text\":\"[ALERT] Kafka Consumer Lag 임계값 초과\\nGroup: $GROUP\\nLag: $TOTAL_LAG\\nThreshold: $ALERT_THRESHOLD\"}"
    fi

    echo ""
done

# 컨슈머 그룹 오프셋 리셋 (주의: 컨슈머가 중지된 상태에서만 실행)
# --to-earliest: 토픽의 가장 처음부터 재생
kafka-consumer-groups.sh \
    --bootstrap-server kafka-broker-1:9092 \
    --group order-read-projection \
    --topic order-events \
    --reset-offsets \
    --to-earliest \
    --execute

# 실행 결과 확인
kafka-consumer-groups.sh \
    --bootstrap-server kafka-broker-1:9092 \
    --describe \
    --group order-read-projection