Saga pattern với Kafka cho giao dịch phân tán: ví dụ đặt tour du lịch

1. Tổng quan

Khi một nghiệp vụ trải qua nhiều microservice, mỗi service có database riêng, bạn không thể bọc tất cả trong một @Transactional. Saga pattern giải quyết bài toán này. Thay vì một giao dịch lớn, bạn chạy một chuỗi giao dịch cục bộ. Mỗi bước có một hành động bù trừ (compensation) để hoàn tác khi bước sau thất bại.

Trong bài này, bạn sẽ cài đặt một saga kiểu orchestration bằng Spring Boot 3 và Kafka cho nghiệp vụ đặt tour du lịch. Đọc xong, bạn sẽ có một orchestrator điều phối các bước, các participant service phản hồi qua Kafka, và một bộ test cho luồng bù trừ.

2. Bài toán đặt tour

Một lượt đặt tour gồm ba bước, mỗi bước thuộc một service:

1. FlightService: giữ chỗ vé máy bay.
2. HotelService: giữ phòng khách sạn.
3. PaymentService: trừ tiền khách hàng.

Giả sử vé và phòng đã giữ xong, nhưng thanh toán thất bại vì thẻ hết hạn mức. Lúc này hệ thống phải hủy phòng rồi hủy vé. Nếu không, khách không trả tiền mà vẫn chiếm chỗ.

Two-phase commit (2PC) có thể giải quyết việc này, nhưng nó khóa tài nguyên trong suốt giao dịch. Nó cũng cần mọi service hỗ trợ XA, và Kafka không tham gia được vào XA. Saga chấp nhận eventual consistency: dữ liệu có thể tạm thời lệch nhau, nhưng cuối cùng sẽ nhất quán.

3. Choreography hay Orchestration

Saga có hai cách tổ chức:

Tiêu chí	Choreography	Orchestration
Điều phối	Mỗi service nghe event của service khác	Một orchestrator gửi command, nhận reply
Luồng nghiệp vụ	Rải rác ở nhiều service	Tập trung một chỗ
Phù hợp	2–3 bước, ít thay đổi	Nhiều bước, có logic bù trừ phức tạp
Rủi ro	Phụ thuộc vòng, khó debug	Orchestrator thành điểm tập trung logic

Với ba bước và một luồng bù trừ ngược chiều, orchestration dễ đọc và dễ test hơn. Bài này dùng cách đó. Các topic Kafka được tổ chức như sau:

tour.commands.flight   ← orchestrator gửi lệnh cho FlightService
tour.commands.hotel    ← orchestrator gửi lệnh cho HotelService
tour.commands.payment  ← orchestrator gửi lệnh cho PaymentService
tour.replies           ← mọi service gửi kết quả về orchestrator

4. Thiết lập dự án

4.1. Dependency

Bạn cần thêm Spring for Apache Kafka. Spring Boot quản lý version giúp bạn:

<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. Cấu hình

Message được gửi dưới dạng JSON. Cấu hình serializer và deserializer như sau:

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 ghi tên class vào header __TypeId__. Phía consumer, JsonDeserializer đọc header đó để biết cần tạo object nào. trusted.packages giới hạn những package được phép deserialize, nhằm tránh lỗ hổng bảo mật.

Tiếp theo, bạn khai báo các topic:

@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. Mô hình message

Bạn dùng record của Java 21 để biểu diễn command và 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) {}

Mỗi message mang bookingId. Giá trị này vừa là định danh của saga, vừa là message key khi gửi lên Kafka. Nhờ vậy, mọi message của cùng một booking rơi vào cùng một partition và được xử lý đúng thứ tự.

6. Định nghĩa saga dưới dạng state machine

Phần quan trọng nhất là logic quyết định bước tiếp theo. Bạn nên tách logic này khỏi Kafka để test được bằng unit test thuần.

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 là một stack lưu các bước đã thành công. Khi cần bù trừ, bạn pop từ đỉnh stack, nên các bước được hoàn tác theo thứ tự ngược lại.

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 đã kết thúc, bỏ qua reply trùng
        }
        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)); // thử hủy lại
        }
        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());
    }
}

Hàm onReply xử lý bốn trường hợp:

• RESERVE thành công: ghi nhận bước đó vào stack. Nếu đây là bước cuối, saga chuyển sang COMPLETED. Nếu chưa, saga gửi lệnh RESERVE cho bước kế tiếp.
• RESERVE thất bại: saga chuyển sang COMPENSATING và bắt đầu hủy bước gần nhất đã thành công.
• CANCEL thành công: saga hủy tiếp bước trước đó. Khi stack rỗng, saga chuyển sang CANCELLED.
• CANCEL thất bại: saga gửi lại lệnh hủy. Bù trừ không được phép bỏ dở.

Lưu ý rằng bước thất bại không nằm trong stack. Nếu thanh toán thất bại, không có gì để hoàn tiền, nên saga chỉ cần hủy phòng và vé.

7. Orchestrator

Orchestrator làm hai việc: gửi command lên Kafka và nghe reply từ 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() dùng bookingId làm key. Tên topic được suy ra từ Step, ví dụ HOTEL thành tour.commands.hotel.

Lưu ý: ConcurrentHashMap chỉ dùng cho demo. Trong production, trạng thái saga phải được lưu vào database. Nếu orchestrator restart giữa chừng, nó cần đọc lại trạng thái để tiếp tục hoặc bù trừ.

8. Participant service

Mỗi participant nghe topic command của mình, thực hiện giao dịch cục bộ, rồi gửi reply. Ví dụ với 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 : "Hết phòng"));
    }

    private boolean reserve(String bookingId) {
        if (reservedRooms.contains(bookingId)) {
            return true; // đã giữ rồi, message bị gửi lại
        }
        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; // hủy thứ chưa từng giữ vẫn tính là thành công
    }
}

Hai chi tiết quan trọng trong đoạn code trên:

• reserve() là idempotent. Kafka đảm bảo giao hàng at-least-once, nên cùng một command có thể đến hai lần. Nếu đã giữ phòng cho bookingId đó, handler trả về thành công mà không trừ phòng lần nữa.
• cancel() cũng idempotent và luôn thành công. Lệnh hủy có thể đến trước lệnh giữ chỗ, hoặc đến cho một booking chưa từng giữ phòng. Cả hai trường hợp đều không được làm hỏng dữ liệu.

FlightCommandHandler và PaymentCommandHandler có cấu trúc tương tự. Riêng với Payment, CANCEL nghĩa là hoàn tiền (refund).

Khi PaymentService từ chối thanh toán, log của orchestrator sẽ như sau:

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

Phòng bị hủy trước, vé máy bay bị hủy sau, đúng thứ tự ngược lại với lúc giữ chỗ.

9. Test luồng saga

Vì TourBookingSaga không phụ thuộc Kafka, bạn test nó bằng JUnit 5 và AssertJ mà không cần broker:

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, "Thẻ bị từ chối")).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);
    }
}

Test thứ hai kiểm tra hành vi quan trọng nhất của saga. Thanh toán thất bại phải dẫn tới hủy khách sạn rồi hủy vé, và saga phải kết thúc ở CANCELLED. Nếu cần test end-to-end, bạn có thể dùng @EmbeddedKafka từ spring-kafka-test hoặc Testcontainers. Phần logic thì nên được test ở mức unit như trên.

10. Lỗi hay gặp

Dual write. Participant ghi database xong rồi mới gửi reply. Nếu service chết giữa hai thao tác đó, saga bị treo. Cách giải quyết chuẩn là Transactional Outbox: ghi reply vào bảng outbox trong cùng transaction với dữ liệu nghiệp vụ. Sau đó một tiến trình riêng (hoặc Debezium CDC) đẩy các bản ghi đó lên Kafka.

Thiếu idempotency. Consumer rebalance hoặc retry sẽ gửi lại message. Mọi handler, kể cả handler bù trừ, phải xử lý trùng an toàn. Ở orchestrator, bạn nên lưu thêm bước đang chờ reply và bỏ qua reply không khớp.

Không dùng key. Nếu gửi message mà không có key, các message của cùng một booking có thể rơi vào nhiều partition khác nhau. Khi đó lệnh CANCEL có thể được xử lý trước lệnh RESERVE.

Thiếu isolation. Saga không có tính Isolation như ACID. Trong lúc saga đang chạy, service khác có thể thấy phòng “đã giữ” dù sau đó phòng bị hủy. Bạn nên dùng trạng thái trung gian như PENDING (semantic lock) và chỉ chuyển sang CONFIRMED khi saga hoàn tất.

Thiếu timeout. Nếu một service không bao giờ trả lời, saga sẽ chờ mãi. Bạn cần một job định kỳ quét các saga quá hạn để bù trừ. Với lệnh CANCEL thất bại liên tục, hãy giới hạn số lần retry và chuyển message sang dead-letter topic để con người xử lý.

Lưu ý: Đừng dùng Saga khi mọi dữ liệu nằm trong cùng một database. Một @Transactional bình thường đơn giản và an toàn hơn nhiều. Saga cũng không phù hợp với nghiệp vụ không chấp nhận trạng thái trung gian, hoặc có bước không thể bù trừ. Ví dụ, một email xác nhận đã gửi thì không thể “thu hồi”. Bạn nên đặt các bước kiểu này ở cuối saga.

11. Kết luận

Trong bài này, bạn đã cài đặt Saga pattern kiểu orchestration với Kafka cho nghiệp vụ đặt tour:

• Logic saga được tách thành một state machine thuần, test được mà không cần broker.
• Orchestrator gửi command và nhận reply qua Kafka, dùng bookingId làm key để giữ thứ tự.
• Các bước bù trừ được thực hiện theo thứ tự ngược lại nhờ một stack.
• Participant xử lý command một cách idempotent.

Khi đưa saga vào production, bạn cần bổ sung ba thứ: lưu trạng thái saga vào database, dùng Transactional Outbox để tránh dual write, và có cơ chế timeout cùng dead-letter topic. Đây là những phần quyết định saga của bạn có thật sự đáng tin cậy hay không.