MSA 분산 트랜잭션과 SAGA 패턴 — 이론부터 실습까지

@Transactional
public void placeOrder(Long userId, long amount) {
    orderRepository.save(new Order(userId, amount)); // 같은 DB
    accountRepository.withdraw(userId, amount);       // 같은 DB
    pointRepository.earn(userId, amount / 100);       // 같은 DB
    // 어느 한 줄이라도 실패하면 세 작업 모두 자동 롤백
}

@Entity
@Table(name = "processed_message")
public class ProcessedMessage {
    @Id
    private String messageId;        // 이벤트의 고유 키 (sagaId + 단계명 등)
    private Instant processedAt;
}

@Service
@RequiredArgsConstructor
public class PaymentService {
    private final ProcessedMessageRepository processed;
    private final AccountRepository accountRepository;

    @Transactional
    public void charge(String sagaId, Long orderId, Long userId, long amount) {
        // 이미 처리한 메시지면 무시 — 같은 saga를 두 번 받아도 1회만 차감
        if (processed.existsById(sagaId)) return;

        Account account = accountRepository.findByUserIdForUpdate(userId);
        account.withdraw(amount);                     // 잔액 부족 시 예외
        processed.save(new ProcessedMessage(sagaId, Instant.now()));
    }
}

public enum OrderStatus { PENDING, APPROVED, REJECTED }

@Entity
@Table(name = "orders")
public class Order {
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private Long userId;
    private long amount;
    private String sagaId;
    @Enumerated(EnumType.STRING)
    private OrderStatus status;      // 사가 완료 전까지 PENDING — 확정으로 취급 금지
    private String rejectReason;

    public static Order create(Long userId, long amount, String sagaId) {
        Order o = new Order();
        o.userId = userId; o.amount = amount; o.sagaId = sagaId;
        o.status = OrderStatus.PENDING;
        return o;
    }
    public void approve()              { this.status = OrderStatus.APPROVED; }
    public void reject(String reason)  { this.status = OrderStatus.REJECTED; this.rejectReason = reason; }
    public boolean canApprove()        { return this.status == OrderStatus.PENDING; }
}

@Transactional
public Long placeOrder(PlaceOrderRequest req) {
    Order order = orderRepository.save(Order.create(...));   // (1) DB 커밋
    // ── 여기서 프로세스가 죽으면? 주문은 남고 이벤트는 영영 안 나간다 ──
    kafka.send("order.created", new OrderCreatedEvent(...));  // (2) 메시지 발행
    return order.getId();
}

@Entity
@Table(name = "outbox")
public class OutboxEvent {
    @Id @UuidGenerator                 // Hibernate 6 UUID 생성 (plain @GeneratedValue 지양)
    private UUID id;
    private String aggregateType;      // "order"  -> Debezium EventRouter 라우팅 키
    private String aggregateId;        // 주문 ID   -> 메시지 key (파티션/순서)
    private String type;               // "OrderCreated"
    @JdbcTypeCode(SqlTypes.JSON)       // 없으면 'jsonb vs varchar' 바인딩 에러로 실제로 터진다
    @Column(columnDefinition = "jsonb")
    private String payload;
    private Instant createdAt;
}

@Transactional
public Long placeOrder(PlaceOrderRequest req) {
    String sagaId = UUID.randomUUID().toString();
    Order order = orderRepository.save(Order.create(req.userId(), req.amount(), sagaId));

    // 같은 트랜잭션 안에서 이벤트도 DB에 적재 — 커밋되면 둘 다, 실패하면 둘 다 무효
    outboxRepository.save(OutboxEvent.of(
        "order", order.getId().toString(), "OrderCreated",
        objectMapper.writeValueAsString(
            new OrderCreatedEvent(sagaId, order.getId(), req.userId(), req.amount()))));
    return order.getId();
}

{
  "name": "outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "saga",
    "plugin.name": "pgoutput",
    "table.include.list": "public.outbox",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.table.field.event.payload": "payload"
  }
}

@Transactional
public Payment refund(String sagaId) {              // 보상 트랜잭션
    Payment p = paymentRepository.findBySagaId(sagaId).orElseThrow();
    if (p.isRefunded()) return p;                     // 멱등 — 두 번 환불 방지
    accountRepository.findByUserIdForUpdate(p.getUserId()).deposit(p.getAmount());
    p.markRefunded();                                 // 차감을 '없애는' 게 아니라 '되돌리는' 기록
    return p;
}

CREATE TABLE account (
    user_id   BIGINT PRIMARY KEY,
    available BIGINT NOT NULL,            -- 사용 가능 잔액
    frozen    BIGINT NOT NULL DEFAULT 0   -- 예약(동결)된 금액
);

CREATE TABLE reservation (
    saga_id    TEXT PRIMARY KEY,          -- 멱등 키
    user_id    BIGINT NOT NULL,
    amount     BIGINT NOT NULL,
    status     TEXT   NOT NULL,           -- RESERVED / CONFIRMED / CANCELLED
    expires_at TIMESTAMPTZ NOT NULL       -- 만료된 예약은 배치로 자동 회수
);

-- Try : 예약 (available -> frozen). 잔액 부족이면 0 rows -> 실패 처리
UPDATE account SET available = available - :amt, frozen = frozen + :amt
WHERE user_id = :uid AND available >= :amt;

-- Confirm : 확정 (frozen 소멸)
UPDATE account SET frozen = frozen - :amt WHERE user_id = :uid;

-- Cancel : 취소 (frozen -> available 복구)
UPDATE account SET available = available + :amt, frozen = frozen - :amt
WHERE user_id = :uid;

services:
  kafka:
    image: apache/kafka:3.7.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: saga
      POSTGRES_PASSWORD: saga
    ports: ["5432:5432"]
    volumes: ["./init:/docker-entrypoint-initdb.d"]   # 스키마 + 시드 데이터

  order-service:    { build: ./order-service,   depends_on: [kafka, postgres], ports: ["8081:8080"] }
  payment-service:  { build: ./payment-service, depends_on: [kafka, postgres], ports: ["8082:8080"] }
  point-service:    { build: ./point-service,   depends_on: [kafka, postgres], ports: ["8083:8080"] }

package com.bedaily.saga.common.event;

// 이벤트 — "이미 일어난 사실"을 알림 (Choreography)
public record OrderCreatedEvent(String sagaId, Long orderId, Long userId, long amount) {}
public record PaymentCompletedEvent(String sagaId, Long orderId, Long paymentId, long amount) {}
public record PaymentFailedEvent(String sagaId, Long orderId, String reason) {}
public record PaymentRefundedEvent(String sagaId, Long orderId) {}

// 커맨드 — "이것을 하라"는 지시 (Orchestration)
public record ChargePaymentCommand(String sagaId, Long orderId, Long userId, long amount) {}
public record RefundPaymentCommand(String sagaId) {}
public record EarnPointCommand(String sagaId, Long orderId, long point) {}

public final class Topics {
    public static final String ORDER_CREATED      = "order.created";
    public static final String PAYMENT_COMPLETED   = "payment.completed";
    public static final String PAYMENT_FAILED      = "payment.failed";
    public static final String PAYMENT_REFUNDED    = "payment.refunded";
    public static final String CMD_CHARGE_PAYMENT  = "cmd.payment.charge";
    public static final String CMD_REFUND_PAYMENT  = "cmd.payment.refund";
    public static final String CMD_EARN_POINT      = "cmd.point.earn";
    private Topics() {}
}

spring:
  kafka:
    bootstrap-servers: kafka:9092
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.bedaily.saga.common.event"
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

@Service
@RequiredArgsConstructor
public class OrderCommandService {
    private final OrderRepository orderRepository;
    private final KafkaTemplate<String, Object> kafka;

    @Transactional
    public Long placeOrder(PlaceOrderRequest req) {
        String sagaId = UUID.randomUUID().toString();
        Order order = orderRepository.save(
                Order.create(req.userId(), req.amount(), sagaId));   // status = PENDING

        // 실무에선 Outbox로 (05장) — 데모 단순화를 위해 직접 발행
        // key=sagaId : 같은 사가는 같은 파티션으로 → 이벤트 순서 보장 + 동시 처리 방지
        kafka.send(Topics.ORDER_CREATED, sagaId,
                new OrderCreatedEvent(sagaId, order.getId(), req.userId(), req.amount()));
        return order.getId();
    }
}

@Component
@RequiredArgsConstructor
public class PaymentEventListener {
    private final PaymentService paymentService;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = Topics.ORDER_CREATED, groupId = "payment-service")
    public void on(OrderCreatedEvent e) {
        try {
            Payment p = paymentService.charge(e.sagaId(), e.orderId(), e.userId(), e.amount());
            kafka.send(Topics.PAYMENT_COMPLETED, e.sagaId(),
                    new PaymentCompletedEvent(e.sagaId(), e.orderId(), p.getId(), e.amount()));
        } catch (InsufficientBalanceException ex) {
            // 실패도 '사실'이므로 이벤트로 알린다 — 주문 서비스가 보고 거절 처리
            kafka.send(Topics.PAYMENT_FAILED, e.sagaId(),
                    new PaymentFailedEvent(e.sagaId(), e.orderId(), ex.getMessage()));
        }
    }
}

@Service
@RequiredArgsConstructor
public class PaymentService {
    private final AccountRepository accountRepository;
    private final PaymentRepository paymentRepository;

    @Transactional
    public Payment charge(String sagaId, Long orderId, Long userId, long amount) {
        // 멱등성 — 같은 saga로 이미 결제했으면 기존 결과 반환
        return paymentRepository.findBySagaId(sagaId).orElseGet(() -> {
            Account account = accountRepository.findByUserIdForUpdate(userId);
            account.withdraw(amount);                       // 잔액 부족 시 예외
            return paymentRepository.save(Payment.completed(sagaId, orderId, userId, amount));
        });
    }
}

@Component
@RequiredArgsConstructor
public class OrderSagaListener {
    private final OrderRepository orderRepository;

    @KafkaListener(topics = Topics.PAYMENT_COMPLETED, groupId = "order-service")
    @Transactional
    public void onPaymentCompleted(PaymentCompletedEvent e) {
        orderRepository.findById(e.orderId()).ifPresent(Order::approve);   // PENDING -> APPROVED
    }

    @KafkaListener(topics = Topics.PAYMENT_FAILED, groupId = "order-service")
    @Transactional
    public void onPaymentFailed(PaymentFailedEvent e) {
        // T1(주문 PENDING)에 대한 보상 = 주문 거절
        orderRepository.findById(e.orderId()).ifPresent(o -> o.reject(e.reason()));
    }
}

@Component
@RequiredArgsConstructor
public class PointEventListener {
    private final PointService pointService;

    // 결제 완료에 얹혀 적립. 실패해도 보상하지 않음 — 재시도(best-effort)만.
    @KafkaListener(topics = Topics.PAYMENT_COMPLETED, groupId = "point-service")
    public void on(PaymentCompletedEvent e) {
        pointService.earn(e.sagaId(), e.orderId(), e.amount() / 100);   // 1% 적립
    }
}

public enum SagaStep { STARTED, PAYMENT_PENDING, COMPLETED, COMPENSATING, REJECTED }

@Entity
@Table(name = "saga_state")
public class SagaState {
    @Id private String sagaId;
    private Long orderId;
    @Enumerated(EnumType.STRING) private SagaStep step;
    @Version private long version;      // 낙관적 락 — 같은 사가 동시 갱신 시 충돌 감지
    private String lastError;
    private Instant updatedAt;
    // start(), markPaymentPending(), complete(), compensating(), reject() ...
}

@Service
@RequiredArgsConstructor
public class OrderSagaOrchestrator {
    private final SagaStateRepository sagaRepo;
    private final OrderRepository orderRepo;
    private final KafkaTemplate<String, Object> kafka;

    @Transactional
    public void start(Long orderId, Long userId, long amount) {
        String sagaId = UUID.randomUUID().toString();
        sagaRepo.save(SagaState.start(sagaId, orderId));        // STARTED -> PAYMENT_PENDING
        kafka.send(Topics.CMD_CHARGE_PAYMENT, sagaId,          // 커맨드도 sagaId 키로
                new ChargePaymentCommand(sagaId, orderId, userId, amount));
    }

    @KafkaListener(topics = Topics.PAYMENT_COMPLETED, groupId = "orchestrator")
    @Transactional
    public void onPaymentCompleted(PaymentCompletedEvent e) {
        SagaState s = sagaRepo.findById(e.sagaId()).orElseThrow();
        // 유실 허용 단계 — fire-and-forget, 응답을 기다리지 않는다
        kafka.send(Topics.CMD_EARN_POINT, e.sagaId(),
                new EarnPointCommand(e.sagaId(), e.orderId(), e.amount() / 100));
        try {
            finalizeOrder(e.orderId());                        // 최종 검증/확정 (재고 확인 등)
            s.complete();                                      // -> COMPLETED
        } catch (OrderFinalizeException ex) {
            // 결제(피벗 이후)는 이미 끝났으므로 되돌릴 수 없다 -> 보상 = 환불 커맨드 발행
            kafka.send(Topics.CMD_REFUND_PAYMENT, e.sagaId(), new RefundPaymentCommand(e.sagaId()));
            orderRepo.findById(e.orderId()).ifPresent(o -> o.reject(ex.getMessage()));
            s.compensating();                                  // -> COMPENSATING
        }
    }

    // 결제 자체가 실패(차감 전) -> 보상할 게 없으므로 주문만 거절
    @KafkaListener(topics = Topics.PAYMENT_FAILED, groupId = "orchestrator")
    @Transactional
    public void onPaymentFailed(PaymentFailedEvent e) {
        SagaState s = sagaRepo.findById(e.sagaId()).orElseThrow();
        orderRepo.findById(e.orderId()).ifPresent(o -> o.reject(e.reason()));
        s.reject(e.reason());                                  // -> REJECTED
    }

    // 환불 완료 통보 -> COMPENSATING 을 REJECTED 로 확정
    @KafkaListener(topics = Topics.PAYMENT_REFUNDED, groupId = "orchestrator")
    @Transactional
    public void onPaymentRefunded(PaymentRefundedEvent e) {
        sagaRepo.findById(e.sagaId()).ifPresent(s -> s.reject("compensated"));
    }

    // 최종 확정 — 검증 실패 시 OrderFinalizeException 으로 보상을 유발
    private void finalizeOrder(Long orderId) {
        Order order = orderRepo.findById(orderId).orElseThrow();
        if (!order.canApprove()) throw new OrderFinalizeException("최종 검증 실패");
        order.approve();                                       // PENDING -> APPROVED
    }
}

@Component
@RequiredArgsConstructor
public class PaymentCommandHandler {
    private final PaymentService paymentService;
    private final KafkaTemplate<String, Object> kafka;

    @KafkaListener(topics = Topics.CMD_CHARGE_PAYMENT, groupId = "payment-service")
    public void onCharge(ChargePaymentCommand c) {
        try {
            Payment p = paymentService.charge(c.sagaId(), c.orderId(), c.userId(), c.amount());
            kafka.send(Topics.PAYMENT_COMPLETED, c.sagaId(),
                    new PaymentCompletedEvent(c.sagaId(), c.orderId(), p.getId(), c.amount()));
        } catch (InsufficientBalanceException ex) {
            kafka.send(Topics.PAYMENT_FAILED, c.sagaId(),
                    new PaymentFailedEvent(c.sagaId(), c.orderId(), ex.getMessage()));
        }
    }

    // 보상 커맨드 — 오케스트레이터가 피벗 이후 실패 시 발행
    @KafkaListener(topics = Topics.CMD_REFUND_PAYMENT, groupId = "payment-service")
    public void onRefund(RefundPaymentCommand c) {
        Payment p = paymentService.refund(c.sagaId());         // 멱등 환불 (07장)
        kafka.send(Topics.PAYMENT_REFUNDED, c.sagaId(),
                new PaymentRefundedEvent(c.sagaId(), p.getOrderId()));
    }
}

docker compose up -d        # kafka + postgres + 3 services

# 시드: user 1 = 잔액 100,000 / user 2 = 잔액 0

# 시나리오 A) 성공 — 잔액 충분
curl -X POST localhost:8081/orders \
  -H 'Content-Type: application/json' \
  -d '{"userId":1,"amount":10000}'
# order.created -> 결제 차감 -> payment.completed
#   -> order APPROVED, point 100 적립

# 시나리오 B) 실패 — 잔액 부족
curl -X POST localhost:8081/orders \
  -H 'Content-Type: application/json' \
  -d '{"userId":2,"amount":5000}'
# order.created -> 잔액 부족 -> payment.failed
#   -> order REJECTED, 차감 없음(보상 불필요), 포인트 미적립

# 결과 확인
curl localhost:8081/orders/1     # { "status": "APPROVED" }
curl localhost:8082/accounts/1   # { "balance": 90000 }
curl localhost:8083/points/1     # { "point": 100 }