Saga パターン: Kafka で実現するマイクロサービスの分散トランザクション

2PC に頼らずに業務一貫性を保つ手法として Saga パターンは定番ですが、実運用で破綻しない設計・実装には Kafka の公式機能理解が欠かせません。

本記事は CCDAK（Confluent Certified Developer for Apache Kafka）出題範囲に寄せて、Kafka のトランザクション API、Exactly-Once セマンティクス、トピック/キー設計、補償設計を実務目線でまとめます。

Saga の基本と Kafka を使う理由

Saga は「一連のローカルトランザクション」と「失敗時の補償アクション」の組合せで最終的整合性を実現します。制御方式はオーケストレーション（中央の調停役が指揮）とコレオグラフィ（各サービスがイベントを購読して自律的に次を起動）の二つ。Kafka は耐久・順序・リプレイ・スケール特性に優れ、どちらの方式でも堅牢な実装を支えます。

CCDAK 観点では、トピック分割とキーでの順序保証、プロデューサの冪等性とトランザクション、コンシューマの sendOffsetsToTransaction、Kafka Streams の exactly_once_v2 が頻出です。設計段階からこれらを織り込むと、障害時の再処理や監査も単純化します。

• オーケストレーション: 中央の Orchestrator がコマンドを配信。可観測性と制御が明快。
• コレオグラフィ: 各サービスがイベントをトリガーに自律進行。結合度を下げやすい。
• Kafka の強み: 追記ログでの順序・耐久、再処理、スケール。補償や監査の根拠を残せる。
• 試験向け要点: idempotence、transactional.id、EOS v2、sendOffsetsToTransaction の正しい使いどころ。

Kafka での Saga（コレオグラフィ）例

イベント/トピック設計とキー戦略

Saga の基本は「事実のイベント」を記録し、それを基に次の行為を決めることです。Kafka ではイベントとコマンドを論理的に分け、キーで順序と局所性を確保します。キーは同一ビジネスエンティティ（例: orderId）で固定し、1 パーティション内での順序性を得ます。

スキーマは後方互換を基本にし、破壊的変更を避けます。Schema Registry を用いるとスキーマ進化を測定可能にでき、CCDAK の出題意図（スキーマ互換性モード）にも沿います。ログ圧縮（compact）を併用するトピックはスナップショット性に向き、Saga の状態投影にも応用できます。

• イベントは過去の事実（OrderCreated、PaymentAuthorized など）を表現する
• コマンドは意図（AuthorizePayment、ReserveInventory など）を表現する
• キーは業務エンティティ ID で一貫させ、順序の破壊を避ける
• 重複起動を見越してイベントにはイベント ID（UUID）を付与
• トピックは用途ごとに分離（-events と -commands を分ける）
• 必要に応じて compact + delete のデュアル保持で最新像と履歴を両立

整合性を担保する基礎: Outbox + 取引的プロデューサ + EOS

サービス内 DB 更新と Kafka 送信の一貫性は Outbox パターンで担保します。アプリは同一ローカルトランザクションでビジネス行と Outbox 行をコミットし、Outbox をポーリングするコンポーネント（または CDC/Connect）が Kafka に配信します。これにより DB と Kafka の間の二相問題を解消できます。

Kafka プロデューサは idempotence を有効化し、必要に応じてトランザクションを使います。トランザクションを用いると複数トピック/パーティションへの書き込みとコンシューマのオフセットコミットを原子的にまとめられます。Kafka Streams は exactly_once_v2 を設定すると内部的にトランザクション制御を行い、入力処理と出力の重複・欠落を抑止します。

• プロデューサ設定: enable.idempotence=true、適切な acks=all、transactional.id を一意に付与
• コンシューマ→プロデューサの処理では sendOffsetsToTransaction を用いて出力とオフセットを同一トランザクションに束ねる
• Kafka Streams の processing.guarantee=exactly_once_v2 は現行の推奨
• Outbox は単純なテーブルに JSON/Avro を格納し、必ず同一 DB トランザクションで書く

失敗時の補償、リトライ、順序制御

Saga では失敗が前提です。各ステップでタイムアウトを設定し、失敗や期限切れをイベントとして事実記録します。補償はビジネス的に可逆なアクションとして設計し、冪等化します（同じ補償を何度実行しても安全）。

リトライは指数バックオフ＋デッドレター（DLQ）で運用可能にします。順序性が重要な場合は、同一キー内での逐次処理と、エラーでの停止が全体をブロックしないように設計します。

• 補償は別コマンド/イベントで明示化（CancelPayment、ReleaseInventory など）
• 補償シーケンスも通常シーケンス同様にイベントソース化して追跡可能にする
• リトライ回数や期限はメッセージヘッダで伝搬させると実装が平易
• DLQ は失敗理由（例外、スタックの要約、元メッセージキー）を付与して可観測性を確保
• 順序維持が必須な処理は同一キー、同一コンシューマグループの 1 並列で処理
• 外部 API 呼び出しは冪等キー（Idempotency-Key）で二重実行に備える

実装例: オーケストレータと Kafka Streams でのステート管理

オーケストレータ方式では、Saga の状態機械をステートストアに保持し、各ステップの結果イベントを受けて次のコマンドを発行します。Kafka Streams を使うと、トピック間のルーティングと状態遷移を 1 トポロジ内で管理でき、exactly_once_v2 により重複や部分失敗を抑制できます。

下記は注文イベントを起点に支払い→在庫→出荷を順次進める最小例です。実運用ではタイムアウト管理、補償遷移、監査ログ、DLQ などを追加します。

• processing.guarantee=exactly_once_v2 を必ず設定
• ステートストアはキー=orderId、値=SAGA 状態（現在ステップ、期限、相関 ID など）
• 外部システム呼び出しはコマンドトピック経由で非同期化し、結果イベントで遷移
• 補償時は専用のコマンドを発行し、成功/失敗もイベントで事実記録

Kafka Streams による簡易オーケストレータ（Java）

Properties p = new Properties();
p.put(StreamsConfig.APPLICATION_ID_CONFIG, "saga-orchestrator");
p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092");
p.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);

StreamsBuilder b = new StreamsBuilder();
StoreBuilder<KeyValueStore<String, String>> store = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("saga-store"),
    Serdes.String(), Serdes.String());
b.addStateStore(store);

KStream<String, OrderEvent> orders = b.stream("order-events",
    Consumed.with(Serdes.String(), orderEventSerde()));

KStream<String, Command> cmds = orders.transformValues(() -> new SagaOrchestrator("saga-store"), "saga-store")
    .flatMapValues((SagaStepResult r) -> r.outgoingCommands());

cmds.split()
    .branch((k, c) -> c.type() == CommandType.AUTH_PAYMENT,
        Named.as("payment"))
    .to("payment-commands", Produced.with(Serdes.String(), commandSerde()));

cmds.split()
    .branch((k, c) -> c.type() == CommandType.RESERVE_STOCK,
        Named.as("inventory"))
    .to("inventory-commands", Produced.with(Serdes.String(), commandSerde()));

// 支払い結果などのイベントを別ストリームで受け、状態を進める
KStream<String, PaymentEvent> pay = b.stream("payment-events",
    Consumed.with(Serdes.String(), paymentEventSerde()));

pay.process(() -> new Processor<String, PaymentEvent>() {
  private KeyValueStore<String, String> kv;
  @Override public void init(ProcessorContext ctx) {
    kv = (KeyValueStore<String, String>) ctx.getStateStore("saga-store");
  }
  @Override public void process(String key, PaymentEvent ev) {
    // 状態更新と次コマンド作成（省略）
  }
}, "saga-store");

KafkaStreams s = new KafkaStreams(b.build(), p);
s.start();

// 注意: Streams は内部でトランザクションを管理し、EOSv2 を実現する

運用・監視と試験対策の要点

Saga は「観測できるか」で成否が決まります。各ステップの開始・成功・失敗・補償をイベントとして残し、相関 ID でたどれるようにします。メトリクスはコンシューマレイテンシ、リトライ率、DLQ 件数、トランザクションの中断率を継続監視します。

CCDAK 対策では、トランザクション境界、オフセットコミットの一貫性、EOS の前提、キーとパーティションでの順序保証、コンパクションと保持ポリシー、スキーマ互換性あたりを確実に押さえます。

• 保持戦略: イベントは delete 保持、最新像や状態投影は compact も検討
• スキーマ互換: 後方互換を既定とし、重大変更は新トピックへ
• 障害復旧: 冪等プロデューサと EOS で重複を抑えつつ、シンク側も冪等化
• トランザクション: 長時間放置による time-out を避け、transactional.id を安定化
• セキュリティ: 監査ログと相関 ID をヘッダに付与、PII はトークナイズ
• テスト: 合成障害（遅延、重複、順序入替）を組み込み、補償の再入可能性を検証

問題で確認

よくある質問