Kafka Streams における Exactly-Once: processing.guarantee と実務設定

Kafka Streams はトランザクションと状態ストアの仕組みにより、入力オフセット・内部状態・出力を書き込み単位で原子的にそろえることができます。これが Exactly-Once Semantics (EOS) です。

本稿では processing.guarantee の正しい選択と、アプリ・ブローカー・トピックの設定観点をまとめます。CCDAK（Confluent Certified Developer for Apache Kafka）で頻出の落とし穴も併記します。

Streams の Exactly-Once とは何か

Kafka Streams の EOS は、1 回の処理サイクルで以下を同一トランザクションに束ねることで、重複や不整合を避ける仕組みです。

1) 入力のコンシューマオフセット送信、2) 内部状態ストアの変更（RocksDB ⇄ changelog）、3) 出力トピックへのレコード書き込み。障害や再起動、リバランス時にも、これらが「全部反映される」か「全部取り消される」かのいずれかになります。

• 対象は Kafka 内で完結する副作用（出力トピック・内部トピック・オフセット）。外部 DB などは対象外。
• EOS は idempotent producer とトランザクションを前提に実現され、Streams では processing.guarantee で切り替えます。
• 最新世代では exactly_once_v2 の利用が推奨。デフォルトは at_least_once のことが多いため明示設定が無難。

processing.guarantee の選択肢と違い

processing.guarantee はモードにより、重複発生の可能性やオフセットコミットの方法、状態ストア整合性が変わります。試験でも設定値と意味の取り違えが狙われます。

実務では、外部システムへ直接副作用を出さない限り exactly_once_v2 を基本選択とし、遅延・スループット要件から at_least_once へ切り替えるかを検討します。

• at_least_once は高スループット・低レイテンシだが重複の可能性あり。アプリ側でデデュープが必要。
• exactly_once_v2 は出力・内部状態・入力オフセットを同一トランザクションで原子的にコミット。重複を防止。
• 設定名はバージョンで変遷があるため、正確には exactly_once_v2 を明示する。

EOS v2 の内部動作とトランザクション境界

EOS v2 では、Streams スレッド単位のプロデューサを用い、出力トピックと内部 changelog、さらに入力オフセット送信を同一トランザクションに含めます。コミット時に全てが原子的に可視化され、アボート時には全てが無効化されます。

リバランスやクラッシュでスレッドが交代しても、プロデューサのフェンシングにより“ゾンビ”からのコミットをブロックし、一貫性を保ちます。

• 入力オフセットは SendOffsetsToTransaction を利用してトランザクションに同梱。
• 状態ストア（RocksDB）の変更は changelog に反映され、復旧では changelog を適用して正確に再構築。
• 単一トランザクションで可視化するため、下流側は部分適用を観測しません。

EOS v2 におけるトランザクション境界（概念図）

設定手順: アプリ・ブローカー・トピックの要点

まずアプリで processing.guarantee=exactly_once_v2 を設定し、application.id は安定（再デプロイで不変）にします。内部トピックの耐障害性を確保するため replication.factor も指定します。

ブローカー側ではトランザクションログのレプリケーションと最小ISRを十分に確保します。トランザクションタイムアウトはブローカー上限内に収めます。

• アプリ設定の要点: processing.guarantee、application.id、内部トピックの replication.factor、適切な commit.interval.ms（EOS では短めでも安全）。
• ブローカー設定の要点: transaction.state.log.replication.factor、transaction.state.log.min.isr をクラスタ規模に合わせる。
• transaction.timeout.ms（プロデューサ側）は broker の transaction.max.timeout.ms 以下に。
• 外部出力は EOS の対象外。必要なら外部側のトランザクション連携、またはアウトボックス/テーブル駆動配信パターンを採用。

Kafka Streams（Java）設定例: Exactly-Once v2

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
// Exactly-Once v2 を明示
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// 内部トピックの耐障害性（StreamsConfig.TOPIC_PREFIX を利用）
props.put(StreamsConfig.TOPIC_PREFIX + "replication.factor", 3);
// トランザクション関連（producer.* で上書き可能）
props.put("producer.transaction.timeout.ms", 600000); // 10 分（broker 上限以下に）
// レイテンシ/スループット調整の一例
props.put("producer.linger.ms", 5);
props.put("producer.batch.size", 32768);
// コミット間隔（EOS では短めでも一貫性は保たれる）
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100);
// スレッド並列度
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> orders = builder.stream("orders");
KTable<String, Long> counts = orders.groupByKey().count();
counts.toStream().to("orders-agg");

KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

運用の勘所とよくある落とし穴

グレースフルシャットダウン時は、現在のトランザクションをコミット/アボートしてから停止する必要があります。SIGTERM ハンドリングや close(Duration) を利用してください。

リバランスではタスクの移動前にトランザクションが片付くまで待機が発生します。スレッド数や入力分割を見直してトランザクション時間を短く保つと安定します。

• 外部DBやREST呼び出しは EOS の外側。二重適用防止には外部のトランザクション/一意制約/アウトボックスを併用。
• 長大な処理バッチは TX タイムアウトを招きやすい。commit.interval.ms やレコード時間当たりの処理量を制御。
• 内部トピックの replication.factor と min.insync.replicas が低いと可用性低下やコミット失敗につながる。

CCDAK 試験対策：覚えておくべきポイント

processing.guarantee=exactly_once_v2 を選ぶと、入力オフセット送信・内部状態更新・出力が同一トランザクションで原子的にコミットされます。外部システムは含まれません。

Streams では transactional.id を自前で付けないのが通常です（内部で生成・管理）。application.id は安定させるのが鉄則です。

• at_least_once は重複の可能性あり。exactly_once_v2 は Kafka 内で重複なし。
• SendOffsetsToTransaction によりオフセットが TX に含まれる点を説明できるように。
• デフォルトが at_least_once のことがある点に注意。明示設定が無難。

問題で確認

よくある質問