Saga pattern과 Kafka를 통한 분산 트랜잭션: 투어 예약 예제

1. 개요

여러 마이크로서비스를 거쳐가는 비즈니스 로직에서, 각 서비스가 자신의 데이터베이스를 가지고 있다면, 모든 것을 하나의 @Transactional로 래핑할 수 없습니다. Saga pattern이 이 문제를 해결합니다. 하나의 큰 트랜잭션 대신 일련의 로컬 트랜잭션을 실행합니다. 각 단계마다 보상 액션(compensation)이 있어서, 뒤의 단계가 실패하면 되돌릴 수 있습니다.

이 글에서는 Spring Boot 3과 Kafka를 사용해 투어 예약 비즈니스 로직을 위한 orchestration 방식의 saga를 구현합니다. 글을 마칠 때쯤이면, 여러 단계를 조율하는 orchestrator, Kafka를 통해 응답하는 여러 참여 서비스, 그리고 보상 흐름을 위한 테스트 코드를 갖게 될 것입니다.

2. 투어 예약 문제

투어 예약은 세 단계로 이루어지며, 각 단계는 하나의 서비스에 속합니다:

1. FlightService: 항공편 좌석을 예약합니다.
2. HotelService: 호텔 객실을 예약합니다.
3. PaymentService: 고객 계정에서 요금을 차감합니다.

항공편과 객실 예약은 성공했지만 신용카드 한도를 초과해서 결제가 실패한다고 가정합시다. 이 경우 시스템은 호텔 예약을 취소한 후 항공편 예약을 취소해야 합니다. 그렇지 않으면 고객은 결제하지 않고도 좌석과 객실을 차지하게 됩니다.

2단계 커밋(2PC)도 이를 해결할 수 있지만, 트랜잭션 내내 리소스를 락합니다. 또한 모든 서비스가 XA를 지원해야 하는데, Kafka는 XA에 참여할 수 없습니다. Saga는 **결과적 일관성(eventual consistency)**을 수용합니다. 데이터가 일시적으로 불일치할 수 있지만, 최종적으로는 일관성을 갖게 됩니다.

3. Choreography 또는 Orchestration

Saga에는 두 가지 조직 방식이 있습니다:

기준	Choreography	Orchestration
조율	각 서비스가 다른 서비스의 이벤트를 감청	하나의 orchestrator가 명령을 보내고 응답을 받음
비즈니스 흐름	여러 서비스에 분산됨	한 곳에 집중됨
적합한 경우	2~3단계, 변경이 적음	많은 단계, 복잡한 보상 로직
리스크	순환 의존성, 디버그 어려움	Orchestrator가 로직 집중점이 됨

3단계와 역순의 보상 흐름의 경우, orchestration이 읽기 쉽고 테스트하기 편합니다. 이 글에서는 이 방식을 사용합니다. Kafka 토픽은 다음과 같이 조직됩니다:

tour.commands.flight   ← orchestrator가 FlightService에 명령을 보냄
tour.commands.hotel    ← orchestrator가 HotelService에 명령을 보냄
tour.commands.payment  ← orchestrator가 PaymentService에 명령을 보냄
tour.replies           ← 모든 서비스가 orchestrator에 결과를 보냄

4. 프로젝트 설정

4.1. Dependency

Spring for Apache Kafka를 추가해야 합니다. Spring Boot가 버전을 관리합니다:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
    <scope>test</scope>
</dependency>

4.2. 설정

메시지는 JSON 형식으로 전송됩니다. 다음과 같이 serializer와 deserializer를 구성합니다:

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    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.nonla.tour

JsonSerializer는 클래스 이름을 __TypeId__ 헤더에 작성합니다. Consumer 쪽에서는 JsonDeserializer가 이 헤더를 읽어 어떤 객체를 생성해야 하는지 알 수 있습니다. trusted.packages는 역직렬화를 허용할 패키지를 제한해서 보안 취약점을 방지합니다.

다음으로, 토픽을 선언합니다:

@Configuration
public class KafkaTopicConfig {

    @Bean
    KafkaAdmin.NewTopics sagaTopics() {
        return new KafkaAdmin.NewTopics(
            TopicBuilder.name("tour.commands.flight").partitions(3).build(),
            TopicBuilder.name("tour.commands.hotel").partitions(3).build(),
            TopicBuilder.name("tour.commands.payment").partitions(3).build(),
            TopicBuilder.name("tour.replies").partitions(3).build());
    }
}

5. 메시지 모델

Java 21의 record를 사용하여 command와 reply를 표현합니다:

package com.nonla.tour;

public enum Step { FLIGHT, HOTEL, PAYMENT }

public enum Action { RESERVE, CANCEL }

public record SagaCommand(String bookingId, Step step, Action action, BigDecimal amount) {}

public record SagaReply(String bookingId, Step step, Action action,
                        boolean success, String reason) {}

각 메시지는 bookingId를 가집니다. 이 값은 saga의 식별자이자 Kafka에 전송할 때의 메시지 키입니다. 따라서 같은 booking의 모든 메시지는 같은 파티션으로 들어가고 올바른 순서로 처리됩니다.

6. State machine으로 saga 정의하기

가장 중요한 부분은 다음 단계를 결정하는 로직입니다. Kafka와 분리하여 순수 단위 테스트로 테스트할 수 있도록 해야 합니다.

public enum SagaStatus { RUNNING, COMPENSATING, COMPLETED, CANCELLED }

public class SagaState {
    private final String bookingId;
    private final BigDecimal amount;
    private final Deque<Step> completed = new ArrayDeque<>();
    private SagaStatus status = SagaStatus.RUNNING;

    public SagaState(String bookingId, BigDecimal amount) {
        this.bookingId = bookingId;
        this.amount = amount;
    }

    public String bookingId() { return bookingId; }
    public BigDecimal amount() { return amount; }
    public Deque<Step> completed() { return completed; }
    public SagaStatus status() { return status; }
    public void status(SagaStatus status) { this.status = status; }
}

completed는 성공한 단계들을 저장하는 스택입니다. 보상이 필요할 때, 스택의 최상단을 팝하므로 단계들은 예약할 때와 반대 순서로 취소됩니다.

public class TourBookingSaga {

    static final List<Step> STEPS = List.of(Step.FLIGHT, Step.HOTEL, Step.PAYMENT);

    public Optional<SagaCommand> start(SagaState s) {
        return Optional.of(command(s, STEPS.getFirst(), Action.RESERVE));
    }

    public Optional<SagaCommand> onReply(SagaState s, SagaReply r) {
        if (s.status() == SagaStatus.COMPLETED || s.status() == SagaStatus.CANCELLED) {
            return Optional.empty(); // saga가 이미 끝났으므로 중복 응답 무시
        }
        if (r.action() == Action.RESERVE && r.success()) {
            s.completed().push(r.step());
            int idx = STEPS.indexOf(r.step());
            if (idx == STEPS.size() - 1) {
                s.status(SagaStatus.COMPLETED);
                return Optional.empty();
            }
            return Optional.of(command(s, STEPS.get(idx + 1), Action.RESERVE));
        }
        if (r.action() == Action.CANCEL && !r.success()) {
            return Optional.of(command(s, r.step(), Action.CANCEL)); // 취소 재시도
        }
        s.status(SagaStatus.COMPENSATING);
        return compensateNext(s);
    }

    private Optional<SagaCommand> compensateNext(SagaState s) {
        if (s.completed().isEmpty()) {
            s.status(SagaStatus.CANCELLED);
            return Optional.empty();
        }
        return Optional.of(command(s, s.completed().pop(), Action.CANCEL));
    }

    private SagaCommand command(SagaState s, Step step, Action action) {
        return new SagaCommand(s.bookingId(), step, action, s.amount());
    }
}

onReply 함수는 네 가지 경우를 처리합니다:

• RESERVE 성공: 해당 단계를 스택에 기록합니다. 마지막 단계라면 saga는 COMPLETED로 전환됩니다. 아니면 다음 단계에 RESERVE 명령을 보냅니다.
• RESERVE 실패: saga는 COMPENSATING으로 전환하고 가장 최근에 성공한 단계를 취소하기 시작합니다.
• CANCEL 성공: saga는 그 앞의 단계를 계속 취소합니다. 스택이 비면 saga는 CANCELLED로 전환됩니다.
• CANCEL 실패: saga는 취소 명령을 다시 보냅니다. 보상은 미완료로 남겨질 수 없습니다.

실패한 단계는 스택에 들어가지 않습니다. 결제가 실패하면 환불할 것이 없으므로, saga는 호텔 예약과 항공편 예약만 취소하면 됩니다.

7. Orchestrator

Orchestrator는 두 가지를 합니다: Kafka에 명령을 보내고 tour.replies에서 응답을 듣습니다.

@Service
public class TourBookingOrchestrator {

    private static final Logger log = LoggerFactory.getLogger(TourBookingOrchestrator.class);

    private final KafkaTemplate<String, Object> kafka;
    private final TourBookingSaga saga = new TourBookingSaga();
    private final Map<String, SagaState> store = new ConcurrentHashMap<>();

    public TourBookingOrchestrator(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    public void book(String bookingId, BigDecimal amount) {
        SagaState state = new SagaState(bookingId, amount);
        store.put(bookingId, state);
        saga.start(state).ifPresent(this::send);
    }

    @KafkaListener(topics = "tour.replies", groupId = "tour-orchestrator")
    public void onReply(SagaReply reply) {
        SagaState state = store.get(reply.bookingId());
        if (state == null) {
            return;
        }
        saga.onReply(state, reply).ifPresent(this::send);
        log.info("booking={} reply={}/{} ok={} -> status={}",
            reply.bookingId(), reply.step(), reply.action(), reply.success(), state.status());
    }

    private void send(SagaCommand cmd) {
        String topic = "tour.commands." + cmd.step().name().toLowerCase();
        kafka.send(topic, cmd.bookingId(), cmd);
    }

    public SagaStatus statusOf(String bookingId) {
        return store.get(bookingId).status();
    }
}

send()는 bookingId를 키로 사용합니다. 토픽 이름은 Step에서 추론되므로, 예를 들어 HOTEL은 tour.commands.hotel이 됩니다.

참고: ConcurrentHashMap은 데모용입니다. 프로덕션 환경에서는 saga 상태를 데이터베이스에 저장해야 합니다. Orchestrator가 중간에 재시작되면, 상태를 읽어서 계속하거나 보상해야 합니다.

8. Participant 서비스

각 participant는 자신의 command 토픽을 듣고, 로컬 트랜잭션을 실행한 후, 응답을 보냅니다. HotelService의 예:

@Component
public class HotelCommandHandler {

    private final KafkaTemplate<String, Object> kafka;
    private final Set<String> reservedRooms = ConcurrentHashMap.newKeySet();
    private final AtomicInteger availableRooms = new AtomicInteger(10);

    public HotelCommandHandler(KafkaTemplate<String, Object> kafka) {
        this.kafka = kafka;
    }

    @KafkaListener(topics = "tour.commands.hotel", groupId = "hotel-service")
    public void handle(SagaCommand cmd) {
        boolean ok = switch (cmd.action()) {
            case RESERVE -> reserve(cmd.bookingId());
            case CANCEL -> cancel(cmd.bookingId());
        };
        kafka.send("tour.replies", cmd.bookingId(),
            new SagaReply(cmd.bookingId(), Step.HOTEL, cmd.action(), ok, ok ? null : "객실 부족"));
    }

    private boolean reserve(String bookingId) {
        if (reservedRooms.contains(bookingId)) {
            return true; // 이미 예약됨, 메시지 재전송됨
        }
        if (availableRooms.getAndUpdate(n -> n > 0 ? n - 1 : n) == 0) {
            return false;
        }
        reservedRooms.add(bookingId);
        return true;
    }

    private boolean cancel(String bookingId) {
        if (reservedRooms.remove(bookingId)) {
            availableRooms.incrementAndGet();
        }
        return true; // 예약한 적 없는 취소도 성공으로 처리
    }
}

위 코드의 두 가지 중요한 세부 사항:

• reserve()는 멱등적입니다. Kafka는 최소 한 번 전달을 보장하므로, 같은 명령이 두 번 올 수 있습니다. 해당 bookingId에 대해 이미 객실을 예약했다면, 핸들러는 성공을 반환하되 객실을 다시 차감하지 않습니다.
• cancel()도 멱등적이며 항상 성공합니다. 취소 명령이 예약 명령보다 먼저 올 수도 있고, 예약한 적 없는 booking에 대해 올 수도 있습니다. 둘 다 데이터를 손상시키지 않습니다.

FlightCommandHandler와 PaymentCommandHandler는 유사한 구조를 가집니다. Payment의 경우, CANCEL은 환불(refund)을 의미합니다.

PaymentService가 결제를 거부할 때, orchestrator의 로그는 다음과 같습니다:

booking=BK-1001 reply=FLIGHT/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=HOTEL/RESERVE ok=true -> status=RUNNING
booking=BK-1001 reply=PAYMENT/RESERVE ok=false -> status=COMPENSATING
booking=BK-1001 reply=HOTEL/CANCEL ok=true -> status=COMPENSATING
booking=BK-1001 reply=FLIGHT/CANCEL ok=true -> status=CANCELLED

호텔이 먼저 취소되고, 항공편이 나중에 취소되며, 이는 예약할 때와 정확히 반대 순서입니다.

9. Saga 흐름 테스트

TourBookingSaga는 Kafka에 의존하지 않으므로, 브로커 없이 JUnit 5와 AssertJ로 테스트할 수 있습니다:

class TourBookingSagaTest {

    private final TourBookingSaga saga = new TourBookingSaga();

    private SagaReply ok(Step step, Action action) {
        return new SagaReply("BK-1", step, action, true, null);
    }

    @Test
    void whenAllStepsSucceed_thenSagaCompleted() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));

        assertThat(saga.start(s)).get().extracting(SagaCommand::step).isEqualTo(Step.FLIGHT);
        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.HOTEL);
        assertThat(saga.onReply(s, ok(Step.HOTEL, Action.RESERVE)))
            .get().extracting(SagaCommand::step).isEqualTo(Step.PAYMENT);
        assertThat(saga.onReply(s, ok(Step.PAYMENT, Action.RESERVE))).isEmpty();

        assertThat(s.status()).isEqualTo(SagaStatus.COMPLETED);
    }

    @Test
    void whenPaymentFails_thenCompensateHotelThenFlight() {
        SagaState s = new SagaState("BK-1", new BigDecimal("15000000"));
        saga.start(s);
        saga.onReply(s, ok(Step.FLIGHT, Action.RESERVE));
        saga.onReply(s, ok(Step.HOTEL, Action.RESERVE));

        SagaCommand first = saga.onReply(s,
            new SagaReply("BK-1", Step.PAYMENT, Action.RESERVE, false, "카드 거부됨")).orElseThrow();
        assertThat(first).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.HOTEL, Action.CANCEL);
        assertThat(s.status()).isEqualTo(SagaStatus.COMPENSATING);

        SagaCommand second = saga.onReply(s, ok(Step.HOTEL, Action.CANCEL)).orElseThrow();
        assertThat(second).extracting(SagaCommand::step, SagaCommand::action)
            .containsExactly(Step.FLIGHT, Action.CANCEL);

        assertThat(saga.onReply(s, ok(Step.FLIGHT, Action.CANCEL))).isEmpty();
        assertThat(s.status()).isEqualTo(SagaStatus.CANCELLED);
    }
}

두 번째 테스트는 saga의 가장 중요한 동작을 확인합니다. 결제 실패는 호텔 취소와 항공편 취소로 이어져야 하며, saga는 CANCELLED로 끝나야 합니다. End-to-end 테스트가 필요하면, spring-kafka-test의 @EmbeddedKafka를 사용하거나 Testcontainers를 사용할 수 있습니다. 로직 부분은 위와 같이 단위 테스트 수준에서 테스트하는 것이 좋습니다.

10. 흔한 실수

Dual write. Participant가 데이터베이스에 쓴 후 응답을 보냅니다. 두 작업 사이에 서비스가 종료되면 saga가 멈춥니다. 표준적인 해결책은 Transactional Outbox입니다: 응답을 비즈니스 데이터와 같은 트랜잭션 내에서 outbox 테이블에 씁니다. 그 후 별도의 프로세스(또는 Debezium CDC)가 이 레코드들을 Kafka로 밀어냅니다.

멱등성 부족. Consumer 리밸런스나 재시도는 메시지를 다시 보냅니다. 보상 핸들러를 포함한 모든 핸들러는 중복을 안전하게 처리해야 합니다. Orchestrator에서는 응답을 기다리는 단계를 추적하고 일치하지 않는 응답을 무시해야 합니다.

키를 사용하지 않음. 키 없이 메시지를 보내면, 같은 booking의 메시지들이 여러 파티션으로 분산될 수 있습니다. 그러면 CANCEL 명령이 RESERVE 명령보다 먼저 처리될 수 있습니다.

격리 부족. Saga는 ACID의 격리(Isolation) 특성이 없습니다. Saga가 실행 중인 동안, 다른 서비스는 나중에 취소될 객실이 “예약됨”으로 보일 수 있습니다. PENDING(semantic lock)과 같은 중간 상태를 사용하고, saga가 완료될 때만 CONFIRMED로 전환해야 합니다.

타임아웃 부족. 한 서비스가 응답하지 않으면, saga는 무한정 기다립니다. 기한을 초과한 saga를 정기적으로 스캔하여 보상하는 작업이 필요합니다. CANCEL 명령이 계속 실패하면, 재시도 횟수를 제한하고 메시지를 dead-letter 토픽으로 이동하여 사람이 처리하도록 해야 합니다.

참고: 모든 데이터가 같은 데이터베이스에 있을 때는 Saga를 사용하지 마세요. 일반적인 @Transactional이 훨씬 간단하고 안전합니다. Saga는 중간 상태를 허용하지 않는 비즈니스 로직이나 보상할 수 없는 단계가 있는 경우에도 적합하지 않습니다. 예를 들어, 발송된 확인 이메일은 “취소”할 수 없습니다. 이러한 단계는 saga의 끝에 배치해야 합니다.

11. 결론

이 글에서 당신은 투어 예약 비즈니스 로직을 위해 Kafka를 사용한 orchestration 방식의 Saga pattern을 구현했습니다:

• Saga 로직은 순수한 state machine으로 분리되어 브로커 없이 테스트할 수 있습니다.
• Orchestrator는 Kafka를 통해 명령을 보내고 응답을 받으며, 순서를 유지하기 위해 bookingId를 키로 사용합니다.
• 보상 단계는 스택 덕분에 역순으로 실행됩니다.
• Participant는 명령을 멱등적으로 처리합니다.

Saga를 프로덕션에 배포할 때는 세 가지를 추가해야 합니다: saga 상태를 데이터베이스에 저장하기, dual write를 피하기 위해 Transactional Outbox 사용하기, 그리고 타임아웃 메커니즘과 dead-letter 토픽 마련하기. 이 부분들이 당신의 saga가 정말 신뢰할 수 있는지를 결정합니다.