kafka-streams-programming

Kafka Streams — 設計、構築、デバッグ

別個のクラスターを必要としない、JVM に組み込んで使うストリーム処理ライブラリです。

⚠️ 重要：参照ファイルは必要なときだけ読み込む

参照ファイルを最初にすべて読まないでください。必要なものだけを、必要なときに読んでください。

• ユーザーが「2 つのトピックをどう結合すればよいですか？」と尋ねた場合 → references/topology-patterns.md の § 結合の判断ツリーだけを読む
• ユーザーが「Kafka Streams アプリを作ってください」と尋ねた場合 → ビルドファイルを作成するときに references/build-templates.md を読む。事前には読まない
• ユーザーが「アプリがクラッシュしています」と尋ねた場合 → その症状に該当する references/debugging.md のセクションを読む
• ほとんどの質問で必要なのは、参照ファイル 10 個すべてではなく、合計 0～2 個です

「念のため」に複数のファイルを先回りして読まないでください

最初に必ず対象環境を確認する

どのモード（Architect、Build、Debug）でも、ユーザーが対象環境を示していない場合は、回答の前に確認してください。Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream。バージョンと認証方式によって推奨内容は変わります。KIP-1071 のサポート、SASL の設定、ACL モデル、トランザクション ID の有効期限、CLI ツール名などは環境ごとに異なります。ユーザーがすでに環境を指定している場合に限り、質問を省略してください。

ユーザーが WarpStream を選択した場合： references/warpstream-optimization.md を読み、標準の設定ベースラインにその上書き設定を適用してください。Kafka Streams に特に影響する点は以下のとおりです。

• EOS はスループットに大きなコストがかかります — exactly_once_v2 は内部で冪等性プロデューサーを有効にします。その結果、同時に処理できるインフライトリクエスト数が限られるため、WarpStream ではスループットが低下します。ユーザーに EOS が強く必要な理由がない限り、at_least_once をデフォルトとし、下流で重複排除を行ってください。
• プロデューサーとコンシューマーの設定は上書きが必要です（バッチサイズを大きくし、linger を長くし、フェッチサイズを大きくします）。共有リファレンスの「Kafka Streams 固有」セクションを参照してください。
• fetch.min.bytes はサポートされていません。設定しないでください。
• replication.factor は形式的な設定です（常に 3）。調整しないでください。
• コストを抑えるには、client.id の末尾に ws_az=<az> を付けるゾーン対応ルーティングが重要です。
• レイテンシは高めです（produce の p50 は約 250ms。Kafka では 1 桁台のミリ秒です）。ユーザーにその点を伝えてください。

モードの判定

ユーザーの意図を判断し、適切なモードに入ってください。

ユーザーの意図	モード	対応
「トピック X のイベントを処理したい……」／「KS アプリを作ってください……」／「集計、フィルタリング、結合をしたい……」	Build	Build モードに進む
「トポロジーをどう設計すればよいですか？」／「KTable と GlobalKTable のどちらを使うべきですか？」／「どの種類の結合が必要ですか？」／「遅延イベントをどう処理すればよいですか？」	Architect	Architect モードに進む
「Streams アプリが停止している／遅い／クラッシュする……」／「なぜリバランスループが発生するのですか？」／「このメトリクスをどう解釈すればよいですか？」	Debug	Debug モードに進む

意図が不明な場合は、コードを生成する前に問題を理解するため、デフォルトで Architect を選んでください。

Flink を優先するかの確認（Confluent Cloud のみ）

ユーザーが KS に触れずに Confluent Cloud の一般的なストリーム処理について尋ねた場合は、代替案として Flink を簡潔に提案してください。長々と説明しないでください。

---

Architect モード

適切なトポロジーを設計します。ユーザーのデータに関する課題を KS のプリミティブに置き換えてください。

ステップ 1：データに関する課題を把握する

まず対象環境を確認してください（前置きの説明を参照）。続いて、以下を尋ねてください（すでに回答済みの場合は省略）。どのデータ（トピック）を扱いますか？ 出力は何ですか？ 入力同士はどのような関係ですか（結合／エンリッチ／グループ化）？

ステップ 2：トポロジーパターンを推奨する

課題に合ったパターンを選んでください（必要なパターンに応じて references/topology-patterns.md の該当箇所だけを読みます）。適合する理由、平易な言葉で説明したデータフロー、使用する KS のプリミティブ、トレードオフと代替案を提示してください。

主な意思決定ツリー

必要に応じて、関連するセクションだけを読んでください。

• トピックを結合する：references/topology-patterns.md § Joins Decision Tree
• 時間単位で集計する：references/topology-patterns.md § Windowing Decision Tree
• エンリッチメント／検索：references/topology-patterns.md § Enrichment Patterns
• Exactly-once：推奨する前に手順を確認してください（下流で重複排除できる場合は、at-least-once のほうがシンプルです）

ユーザーの確認後、ビルドモードに進みます。

---

ビルドモード

完全で実行可能な Kafka Streams プロジェクトを生成します。

ステップ 1：要件を確認する

質問してください（すでに回答済みの場合はスキップします）。

1. アプリの機能：曖昧な要望には、具体化を促してください。
2. トピックとデータフロー：入力／出力トピックはありますか？スキーマは定義済みですか？定義済みの場合は、スキーマを取得してください（新規作成しないでください）。トピックはどのようにつながりますか（結合／検索／独立）？
3. スキーマ形式（既存のものを使用する場合はスキップ）：Avro（デフォルト）｜Protobuf｜JSON Schema
4. ビルドツール：Gradle（デフォルト）｜Maven
5. 対象環境（必須）：Apache Kafka｜Confluent Platform｜Confluent Cloud｜WarpStream（設定を生成する際は references/config-baseline.md を読みます。WarpStream の場合は、クライアントのオーバーライドについて references/warpstream-optimization.md も読みます）
6. 認証情報：CC には 2 つの API キー（Kafka + SR）が必要です。CP/AK にはブートストラップ URL と SR URL、および認証タイプが必要です（必要に応じて references/cli-commands.md を読みます）
7. デプロイ規模：パーティション数は？インスタンス数は？状態サイズは？（必要に応じて references/architecture.md または references/production-hardening.md § Deployment Sizing を読みます）
8. テストデータ：データはありますか？それともサンプルデータの生成を希望しますか？

ステップ 2：リソースを計画する

計画を提示します。作成するトピック（ソース／出力／DLQ）と登録するスキーマを示してください。変更ログ／再パーティション用トピックは KS が自動作成します。入力トピックがすでに存在するとユーザーが述べた場合は、create-topics.sh から除外してください。このスクリプトでは新規トピック（通常は出力 + DLQ）のみを作成します。

ステップ 3：プロジェクトを生成する

以下を生成します：プロジェクト構成、スキーマ、App.java、TopologyBuilder.java、設定、simplelogger.properties、docker-compose（ローカル環境の場合）、スクリプト、TopologyTest.java、.env.example、監視に関するコメント。

必要に応じて参考資料を読みます。

• トポロジーのコード？ → 該当するパターンについて references/topology-patterns.md
• ビルドファイルの構成？ → references/build-templates.md
• スキーマの構文？ → references/schema-patterns.md
• 設定プロパティ？ → 環境固有のブロックについて references/config-baseline.md
• スクリプト？ → scripts/create-topics.sh、scripts/teardown.sh
• ローカル開発？ → references/docker-compose.md

Gradle：ビルドファイルの作成後、gradle wrapper --gradle-version 8.12 を実行します。

ユーザーがサンプルデータを希望する場合は、SampleDataProducer.java と produce タスクを生成します。

ステップ 4：本番環境向けの強化（対象が本番環境の場合）

**条件：**ユーザーが「production」「prod」「deploy」と述べる場合、K8s／ECS／Docker Swarm を指定する場合、または複数インスタンスを要求する場合。

本番環境向けコンポーネントを追加します（必要に応じて詳細は references/production-hardening.md を読みます）：Logback JSON ロギング、logback.xml、ヘルスチェックエンドポイント、JVM チューニングを施した Dockerfile、KIP-1034 DLQ ハンドラー、K8s YAML（K8s の場合）、shadow／fat jar プラグイン。

ステップ 5：コードを説明する

トポロジー、設定の選択、実行方法、監視項目を説明します。group.protocol=streams（KIP-1071）により、リバランスが 50〜80％高速化することに触れてください（AK 4.2+／CP 8.2+ が必要です）。

ステップ 6：引き渡し前にアプリを実行する

タスクを完了したと宣言する前に、実際のブローカーに接続してアプリを起動し、RUNNING に到達することを確認してください。 コードを生成してコンパイルが成功し、TopologyTestDriver のテストに合格していても、起動時に失敗することがあります。バージョン不一致による NoClassDefFoundError、気付かれないロガーへのフォールバック、実行時依存関係の不足、インポートパスのエラーなどは、compile と test だけでは見つからず、実際のブローカー／Schema Registry に接続したときに初めて表面化します。ビルドが成功しても、アプリが動作するとは限りません。

ステップ 1 で選択した対象環境に応じて分岐します。

ローカル（生成された docker-compose.yml を使った Apache Kafka または Confluent Platform）：

1. docker compose up -d を実行し、Kafka と SR が正常になるまで待ちます（docker compose ps、または SR の /subjects に curl）
2. ./create-topics.sh を実行します
3. アプリをバックグラウンドで起動し（./gradlew run または mvn exec:java）、実行中のログを確認できるようにします
4. ログを追跡し、約 30 秒以内に State transition from REBALANCING to RUNNING が表示されることを確認します。表示されない場合は、実際のスタックトレースを確認し、references/debugging.md § Startup Failures を参照して原因を特定し、修正して再起動し、再確認します
5. ユーザーがサンプルデータを希望している場合は、数件のレコードを生成し、出力先トピックに出力が現れることを確認します
6. アプリを停止し、docker compose down を実行します（ユーザーが反復作業のために起動したままにしたい場合は、起動したままにします。ユーザーに確認してください）

Confluent Cloud： クラスターと SR API キーはユーザーが所有しているため、通常は自分でエンドツーエンドの実行を行えません。可能な範囲で実施してから、残りを引き渡します。

1. .env に実際の CC 認証情報がある場合（ユーザーが実際の .env を設定済みの場合）：CC を接続先としてアプリをローカルで実行し（./gradlew run は .env を自動で読み込みます）、上記の手順 3～5 に従います。CC だからという理由だけで省略しないでください。認証情報があれば実行します。
2. 認証情報がプレースホルダー、または提供されていない場合：成功した実行を捏造してはいけません。代わりに、以下を行います。
   • ./gradlew build（コンパイル＋単体テスト）を実行し、結果を報告します
   • ユーザーが確認のために実行する必要がある正確なコマンド（./create-topics.sh --cloud、./gradlew run、references/verification.md § Confluent Cloud に記載されているコンシュームコマンド）と、成功時の状態（State transition from REBALANCING to RUNNING、出力先トピック上のレコード）を列挙します
   • ユーザーに「API キーがないため、お客様の CC クラスターに対して実行できませんでした。上記の手順を実行し、エラーがあれば貼り付けてください」と明確に伝えます

WarpStream： WarpStream クラスターと認証情報はユーザーが所有しているため、通常は自分でエンドツーエンドの実行を行えません。Confluent Cloud と同じ方法で対応します。

1. .env に実際の WarpStream 認証情報がある場合：アプリをローカルで実行し（./gradlew run は .env を自動で読み込みます）、ローカルの手順 3～5 に従います。WarpStream はメタデータのレイテンシーが高いため、State transition from REBALANCING to RUNNING の表示に時間がかかる場合があることに注意してください。
2. 認証情報がプレースホルダー、または提供されていない場合：成功した実行を捏造してはいけません。代わりに、以下を行います。
   • ./gradlew build（コンパイル＋単体テスト）を実行し、結果を報告します
   • ユーザーが確認のために実行する必要がある正確なコマンド（./create-topics.sh、./gradlew run）と、成功時の状態（State transition from REBALANCING to RUNNING、出力先トピック上のレコード）を列挙します
   • ゾーンを考慮したルーティングのために、.env で client.id に ws_az=<az> を設定するようユーザーに伝えます
   • ユーザーに「認証情報がないため、お客様の WarpStream クラスターに対して実行できませんでした。上記の手順を実行し、エラーがあれば貼り付けてください」と明確に伝えます

引き渡し時には、上記のどの作業を行ったかを明確に伝えてください。実行して RUNNING を確認した場合は、そのように伝えます。コンパイルのみの場合は、そのことだけを伝えます。実施していない実行時検証を行ったかのように示してはいけません。

CC のコンシュームコマンド、スキーマ対応プロデューサー、リセット手順については、references/verification.md を参照してください。

---

デバッグモード

ステップ 1：問題を分類する

症状	カテゴリ	参照先
アプリの起動時にクラッシュする	起動失敗	references/debugging.md § Startup Failures
アプリは動作するが、出力がない／処理が停止する	処理の停止	references/debugging.md § Processing Stalls
リバランスが繰り返される／常にリバランスしている	リバランス	references/debugging.md § Rebalancing Issues
ラグが大きい／処理が遅い	パフォーマンス	references/debugging.md § Performance
デシリアライズエラー／ポイズンピル	データ品質	references/debugging.md § Deserialization Errors
ステートストアの問題（破損、増大、復旧）	状態	references/debugging.md § State Store Issues
スレッドの障害／StreamsUncaughtExceptionHandler	スレッドの健全性	references/debugging.md § Thread Failures
メモリの問題（OOM、高いヒープ使用量、RocksDB）	メモリ	references/debugging.md § Memory Issues
WarpStream でスループットが低い、または KAFKA_STORAGE_ERROR が発生する	WarpStream の設定	references/warpstream-optimization.md

ステップ 2：コンテキストを収集する

まず対象環境を確認してください（前文を参照）。デバッグの手順は、環境によって分岐することがほとんどです。次に、エラーメッセージ、設定、KS／Java のバージョン、新規アプリかデグレードかを確認してください。

ステップ 3：診断して修正する

特定したカテゴリに該当する references/debugging.md のセクションを読みます。説明を添えて修正方法を提示してください。

---

不変条件チェックリスト

必ず適用する基本設定です。実装の詳細が必要な場合にのみ、リファレンスファイルを読んでください。

1. スキーマ化されたデータ：SR の SerDe（SpecificAvroSerde、KafkaProtobufSerde、KafkaJsonSchemaSerde）を使用します。schema.registry.url、default.key.serde、default.value.serde を設定します。JSON Schema の場合は、json.value.type を設定します。Protobuf の場合は、specific.protobuf.value.type を設定します（references/config-baseline.md）
2. バージョン：KS 4.x／CP 8.x、Java 17+
3. KIP-1071：group.protocol=streams（デフォルト）を使用します。UnsupportedVersionException が発生する場合は削除します。非対応機能：静的メンバーシップ、正規表現トピック、スタンバイレプリカ、ウォームアップレプリカ（references/topology-patterns.md § Assignment Strategy）
4. 4 層のエラーハンドリング：DeserializationExceptionHandler、ProcessingExceptionHandler（KIP-1034）、ProductionExceptionHandler、StreamsUncaughtExceptionHandler を使用します。未捕捉例外には MaxFailures パターンを使用します（references/production-hardening.md § Error Handling）
5. 明示的な命名：ensure.explicit.internal.resource.naming=true を設定します
6. グレースフルシャットダウン：SIGTERM／SIGINT で streams.close(30s) を呼び出すフックを設定します
7. モニタリング：metrics.recording.level=INFO を設定します（references/config-baseline.md）
8. ログの詳細度：simplelogger.properties を生成します（references/build-templates.md）
9. 防御的なトポロジー：本番環境ではラムダ式を保護します（null チェック、try/catch）。開発環境では、より簡潔な実装でも問題ありません。
10. スキーマの同等性：Avro／Protobuf／JSON Schema のすべてに対応しています（references/build-templates.md、references/schema-patterns.md）
11. テストのキャッシュ：TopologyTestDriver のテストでは、非決定的なアサーションを避けるため、statestore.cache.max.bytes=0 が必要です。
12. Avro の論理型の Java マッピング：Avro 1.12+ では、timestamp-millis／timestamp-micros に対して java.time.Instant、date に対して LocalDate、decimal に対して BigDecimal などを生成します。生成されたセッターメソッドに long／int のリテラルを直接渡してはいけません。Instant.EPOCH、Instant.now()、Instant.ofEpochMilli(...) を使用してください。タイムスタンプの比較では、Math.max()／Math.min() ではなく Instant.isAfter()／isBefore() を使用してください。トポロジーのコード、集約の初期化処理、プロデューサー、およびテストヘルパーに適用されます（references/schema-patterns.md § Java type mapping）。

---

同梱スクリプト

scripts/：create-topics.sh（トピックを事前作成、--cloud）、teardown.sh（トピック／状態を削除、--cloud）、produce-test-data.sh（リクエストされた場合に生成）。

参照ファイル（必要に応じて読み込んでください）

references/topology-patterns.md — 設計、結合、ウィンドウ、集約 | references/architecture.md — 内部構造、サイジング | references/debugging.md — トラブルシューティング | references/config-baseline.md — 設定 | references/build-templates.md — プロジェクト構成 | references/schema-patterns.md — Avro/Protobuf/JSON | references/production-hardening.md — 本番環境のセットアップ | references/cli-commands.md — CLI | references/docker-compose.md — ローカル開発 | references/verification.md — チェックリスト | references/warpstream-optimization.md — WarpStream クライアント設定のオーバーライド