JA
日本語EspañolFrançaisPortuguêsTiếng ViệtEnglish简体中文한국어Deutsch
ホーム / アーキテクチャ・設計 / Kafka

KafkaでのSagaパターンによる分散トランザクション：ツアー予約の例

Cris The Dev · 2026/10/12 · 10 分で読めます · 98

1. 概要

複数のマイクロサービスにまたがるビジネスプロセスでは、各サービスが独立したデータベースを持つため、すべてを1つの@Transactionalでラップすることはできません。Sagaパターンはこの問題を解決します。単一の大きなトランザクションの代わりに、ローカルトランザクションのシーケンスを実行します。各ステップには補償アクション(compensation)があり、後続のステップが失敗したときにロールバックできます。

このチュートリアルでは、ツアー予約ビジネスのためにSpring Boot 3とKafkaを使用してorchestration型のsagaを実装します。完了後、複数のステップを調整するオーケストレーター、Kafkaを通じて応答するparticipantサービス、補償フローをテストするテストスイートを手に入れます。

2. ツアー予約の問題

ツアー予約は3つのステップで構成され、各ステップは異なるサービスに属します：

  1. FlightService：飛行機チケットの座席確保。
  2. HotelService：ホテルの部屋確保。
  3. PaymentService：顧客からの支払い処理。

チケットと部屋の確保が完了したが、支払いが失敗したと仮定します。カードの利用限度額に達しているとします。この場合、システムは部屋をキャンセルし、次にチケットをキャンセルする必要があります。そうしないと、顧客は支払わないまま座席を占有することになります。

2フェーズコミット(2PC)はこれを解決できますが、トランザクション中にリソースをロックしています。またすべてのサービスがXAをサポートする必要があり、KafkaはXAに参加できません。Sagaは**結果的一貫性(eventual consistency)**を受け入れます。つまり、データは一時的に矛盾する可能性がありますが、最終的には一貫性を保ちます。

3. ChoreographyかOrchestration

Sagaには2つの組織方法があります：

基準 Choreography Orchestration
調整 各サービスが他のサービスのイベントをリッスン 1つのオーケストレーターがコマンドを送信し、応答を受け取る
ビジネスフロー 複数のサービスに分散 1か所に集約
適用場面 2～3ステップ、変更が少ない 多くのステップ、複雑な補償ロジック
リスク 循環依存、デバッグが困難 オーケストレーターが論理の集約点となる

3つのステップと逆順の補償フローを持つこの場合、orchestration型の方が読みやすくテストしやすいです。このチュートリアルではこの方法を使用します。Kafkaトピックは以下のように組織されています：

tour.commands.flight   ← オーケストレーターがFlightServiceにコマンドを送信
tour.commands.hotel    ← オーケストレーターがHotelServiceにコマンドを送信
tour.commands.payment  ← オーケストレーターがPaymentServiceにコマンドを送信
tour.replies           ← すべてのサービスが結果をオーケストレーターに送信

4. プロジェクトのセットアップ

4.1. 依存関係

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__ヘッダーに書き込みます。消費側では、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を使ってコマンドと応答を表現します：

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に送信するときのメッセージキーでもあります。これにより、同じ予約のすべてのメッセージが同じパーティションに入り、正しい順序で処理されます。

6. sagaをstate machineとして定義

最も重要な部分は、次のステップを決定するロジックです。このロジックを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は成功したステップを記録するスタックです。補償が必要な場合、スタックの上部からpopするため、ステップは予約時と逆順でロールバックされます。

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関数は4つのケースを処理します：

  • RESERVEが成功：そのステップをスタックに記録します。最後のステップの場合、sagaはCOMPLETEDに移行します。そうでない場合、sagaは次のステップに対してRESERVEコマンドを送信します。
  • RESERVEが失敗：sagaはCOMPENSATINGに移行し、最も最近成功したステップのキャンセルを開始します。
  • CANCELが成功：sagaは前のステップをキャンセルします。スタックが空になると、sagaはCANCELLEDに移行します。
  • CANCELが失敗：sagaはキャンセルコマンドを再送信します。補償は中止することが許されていません。

失敗したステップはスタックに存在しないことに注意してください。支払いが失敗した場合、払い戻すものはないため、sagaは部屋とチケットをキャンセルするだけです。

7. オーケストレーター

オーケストレーターは2つのことを行います：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の状態をデータベースに保存する必要があります。オーケストレーターが途中で再起動された場合、状態を読み直して続行または補償する必要があります。

8. Participantサービス

各participantはそれぞれのコマンドトピックをリッスンし、ローカルトランザクションを実行して、応答を送信します。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; // 予約されたことのない部屋のキャンセルも成功として扱われます
    }
}

上記のコードには2つの重要な詳細があります：

  • reserve()は**べき等(idempotent)**です。Kafkaは最低1回の配信を保証するため、同じコマンドが2回到着する可能性があります。そのbookingIdの部屋が既に予約されている場合、ハンドラーは再び部屋数を減らさずに成功を返します。
  • cancel()もべき等で、常に成功します。キャンセルコマンドは予約コマンドより前に到着する可能性があり、予約されたことのない予約に対して到着する可能性があります。どちらの場合もデータを破損させてはいけません。

FlightCommandHandlerとPaymentCommandHandlerは同じ構造です。Paymentの場合、CANCELは払い戻し(refund)を意味します。

PaymentServiceが支払いを拒否すると、オーケストレーターのログは以下のようになります：

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);
    }
}

2番目のテストはsagaの最も重要な動作をチェックします。支払いが失敗し、ホテルをキャンセルしてから飛行機チケットをキャンセルする必要があり、sagaはCANCELLEDで終了する必要があります。end-to-endテストが必要な場合は、spring-kafka-testからの@EmbeddedKafkaまたはTestcontainersを使用できます。ロジック部分は上記のようなユニットレベルでテストするべきです。

10. 一般的なエラー

Dual write。 Participantがデータベースに書き込んだ後、応答を送信します。2つのアクション間でサービスが停止した場合、sagaがハング状態になります。標準的な解決策はTransactional Outboxです：応答をoutboxテーブルに書き込みます（ビジネスデータと同じトランザクション内）。その後、別のプロセス（またはDebezium CDC）がそれらのレコードをKafkaに送信します。

べき等性の欠如。 Consumer rebalanceまたはretryはメッセージを再送信します。ハンドラー（補償ハンドラーを含む）は重複を安全に処理する必要があります。オーケストレーターでは、応答待ちステップを追加で保存し、マッチしない応答を無視すべきです。

キーを使用しない。 キーなしでメッセージを送信すると、同じ予約のメッセージが異なるパーティションに入る可能性があります。その場合、CANCELコマンドはRESERVEコマンドより前に処理される可能性があります。

isolationの欠如。 Sagaは通常のACIDのようなIsolationプロパティを持ちません。sagaが実行中、別のサービスは部屋が「予約済み」と見える可能性があり、後で部屋がキャンセルされます。PENDING（semantic lock）のような中間状態を使用し、sagaが完了した場合にのみCONFIRMEDに移行することをお勧めします。

タイムアウトの欠如。 サービスが応答しない場合、sagaは永遠に待機します。期限超過のsagaをスキャンして補償するために、定期的なジョブが必要です。CANCELコマンドが継続的に失敗する場合、再試行回数を制限し、メッセージをdead-letter topicに移動して人間が処理できるようにします。

注意： すべてのデータが同じデータベースにある場合はSagaを使用しないでください。通常の@Transactionalの方がはるかにシンプルで安全です。Sagaは中間状態を受け入れることができない、またはロールバック不可能なステップがあるビジネスプロセスにも適していません。例えば、確認メールが送信されたら「取り消す」ことはできません。このようなステップはsagaの最後に配置する必要があります。

11. 結論

このチュートリアルでは、ツアー予約ビジネスのためにKafkaを使用してorchestration型のSagaパターンを実装しました：

  • sagaロジックは純粋なstate machineに分離され、ブローカーなしでテストできます。
  • オーケストレーターがKafkaを通じてコマンドを送信し応答を受け取り、bookingIdをキーとして使用して順序を保証します。
  • 補償ステップはスタックのおかげで逆順で実行されます。
  • Participantはコマンドをべき等な方法で処理します。

sagaを本番環境にデプロイする場合、3つのことを追加する必要があります：sagaの状態をデータベースに保存する、dual writeを避けるために Transactional Outboxを使用する、timeoutとdead-letter topicメカニズムを持つことです。これらはsagaが実装する部分であり、本当に信頼できるかどうかを決定します。

kafkasagamicroservicesspring-bootdistributed-transaction
著者のアプリHoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

同じトピックの記事

Kafka · 2026/10/12 · 8 分で読めます · 312

Change Data CaptureとDebeziumとKafka

Debeziumを使ってPostgreSQLのWALを読み込み、すべてのデータ変更をKafkaにプッシュし、Spring Bootで消費。データ書き込みコードの変更は不要です。