developing-kafka-java-client

質問 #1（既存のアプリか、新規作成か）、#2（対象環境）、#3（プロデューサー、コンシューマー、または両方）について明示的に質問し、回答を受け取るまでは、コードを生成したり、プロジェクトをスキャフォールドしたり、ファイルを変更したりしないでください。ユーザーのプロンプトで一部の質問に回答されている場合も、生成する前に理解した内容を確認してください。 これは、どれほど具体的に見えるプロンプトであっても、すべてのプロンプトに適用されます。

まず、「このプロジェクトを進めるにあたり、Confluent Kafka Java Client スキルを使用します。」と伝えてください。

Confluent Kafka Java Client の作成

Apache Kafka Java クライアント（KafkaProducer／KafkaConsumer）と Confluent Schema Registry シリアライザーを使用して、Kafka にメッセージを送信する、または Kafka からメッセージを受信する、本番環境に対応した Java プロジェクトを生成します。対象環境は Confluent Cloud（マネージド）、Local Docker、WarpStream（Kafka 互換で、オブジェクトストレージを基盤とする）の 3 種類です。ビルドツールは Maven（デフォルト）と Gradle、スキーマ形式は Avro（デフォルト）、JSON Schema、Protobuf の 3 種類に対応しています。生成されるコードは、Confluent のベストプラクティスに従います。

ステップ 1：要件の確認

コードを生成する前に、以下の質問を確認してください。ユーザーのプロンプトですでに明示的に回答されている質問は省略してください。形式的に聞き直す必要はありません。たとえば、「Gradle を使い、Confluent Cloud 上で同期送信するプロデューサーとコンシューマーを作成してください」という依頼では、#2、#3、#4、#5 が回答済みです。残るのは #1、#6、#7、#8、#9、#10 のみです。

必須の確認ゲート — ユーザーがすべての質問に回答済みであっても、省略してはいけません。 ファイルを作成する前に、必ず次の内容を含むメッセージを 1 回送信してください。

1. 抽出した回答を、短い箇条書きで要約します（例：「対象：Confluent Cloud · コンポーネント：プロデューサー + コンシューマー · 送信パターン：非同期コールバック · ビルド：Maven · スキーマ：Avro · 新規作成：はい」）。
2. 未回答の質問があれば、その場で尋ねます。
3. 作業を進める前に、内容の確認または訂正を明示的に依頼します。

その後は停止し、ユーザーの返答を待ってください。要約を送るターン内でファイルを生成したり、処理を進めたりしてはいけません。また、すべての情報が指定されたプロンプトであっても、ただちに生成することへの同意だと見なしてはいけません。プロンプトの誤解を防ぐための要約は、#1～#10 のすべてが事前に回答されている場合にも必要です。この確認ゲートを省略できるのは、この会話ですでにユーザーが要約を確認している場合に限られます。

#1、#2、#3 の回答を推測してはいけません。いずれかがプロンプトで回答されていない場合は、質問してください。

1. 既存のアプリケーションに Kafka を追加しますか、それともゼロから始めますか？
   • ユーザーが既存の Java コードを持っている場合（既存のプロジェクトに言及している、pom.xml／build.gradle がある、Spring Boot／Quarkus／Micronaut を使用しているなど）は、新しいプロジェクトをひな形から作成しないでください。代わりに、（a）既存のプロデューサーまたはデータ送信コードを特定し、（b）Schema Registry にすでにスキーマを登録しているか確認し、（c）リファレンスファイルのパターンに従って、既存コードに Schema Registry との連携を追加します。不足しているファイルだけを生成し（例：KafkaConfig.java、src/main/avro/value.avsc）、既存のコードを直接変更してください。既存のビルドファイルに Confluent Maven リポジトリとシリアライザーの依存関係を追加します。
   • ユーザーが Schema Registry を使わずに Kafka にすでにデータを送信している場合（例：StringSerializer と独自実装の JSON を使用している）、移行を支援してください：（1）既存のメッセージ構造から Avro／JSON スキーマを生成する、（2）スキーマを登録する、（3）生の String／byte[] シリアライザーを Confluent Schema Registry シリアライザーに置き換える。既存のコードは破棄しないでください。
   • ゼロから始める場合は、以下の手順で全体をひな形から作成します。
2. 対象環境はどれですか？ — Confluent Cloud、ローカル Kafka（Docker）、または WarpStream。ユーザーが言及していない場合も、必ず確認してください。「ローカル」、「docker」、「セルフホスト可能」と言及している場合や、クラウドアカウントなしで Kafka を試したいだけの場合は、ローカル Docker を選択します。「Confluent Cloud」、「CC」と言及している場合や、既存のクラウド認証情報がある場合は、Confluent Cloud を選択します。「WarpStream」と言及している場合は、WarpStream を選択します。希望がないと確認できた場合は Confluent Cloud をデフォルトにしますが、必ず最初に確認してください。
   • WarpStream の場合： references/warpstream-optimization.md を読み、そのリファレンスにある Java クライアントのオーバーライドを適用します。主な変更点：冪等性を無効にし、バッチサイズとインフライトリクエスト数を大幅に増やし、大きなフェッチサイズを設定し、ゾーン認識ルーティングのために client.id に ws_az=<az> を追加します。エンティティ単位の順序付けが必要な場合を除き、スティッキーパーティショニングには null のメッセージキーを推奨します。
3. プロデューサー、コンシューマー、またはその両方ですか？
4. 送信パターンはどれですか？（プロデューサーを使う場合のみ。）Java の KafkaProducer.send() は常に Future<RecordMetadata> を返します。つまり、プロデューサーは本質的に非同期です。選択するのは、結果の処理方法です。ユーザーが選べるよう支援してください。
   • コールバックを使った非同期処理（推奨のデフォルト）：producer.send(record, callback)。ノンブロッキングです。Callback がレコードごとの配信結果とエラーを処理します。ほとんどのアプリケーションで最も高いスループットが得られます。references/AvroProducer.java を使用します。
   • 同期処理：producer.send(record).get()。ブローカーが各レコードを確認応答するまでブロックします。スループットは最も低くなりますが、メッセージごとの確認が最も確実で、エラー処理も最もシンプルです。バッチ／ETL スクリプトや、次に進む前に書き込みを確実に完了させる必要があるステップに適しています。references/AvroProducerSync.java を使用します。
   • （ファイア・アンド・フォーゲット — コールバックも .get() も指定せずに producer.send(record) を呼び出す方法 — は説明のために挙げていますが、推奨しません。配信エラーが黙って無視されるためです。） ユーザーがバッチ、ETL、または「書き込みごとに確認する」と言及した場合は、同期処理をデフォルトにします。それ以外の場合は、コールバックを使った非同期処理をデフォルトにします。
5. ビルドツールはどれですか？ — Maven（デフォルト、pom.xml）または Gradle（build.gradle）。Gradle、既存の build.gradle、または Kotlin DSL に言及している場合は、Gradle（references/build.gradle）を使用します。それ以外の場合は Maven（references/pom.xml）をデフォルトにします。
6. 使用したい既存のスキーマはありますか？ ある場合は、貼り付けるかファイルパスを指定するようユーザーに依頼し、生成する代わりにそのスキーマを使用します。ない場合は、データのフィールドについて質問します。
7. どのようなデータを生成しますか？（既存のスキーマがない場合のみ。対応するスキーマとサンプルデータを生成できるよう、フィールド名と型を確認します。）
8. トピック名は何ですか？（デフォルト：demo-topic）
9. コンシューマーグループ ID は何ですか？（コンシューマーを使う場合のみ。デフォルト：java-consumer-group）
10. Consumer API と Share Consumer API のどちらを使いますか？（コンシューマーを使う場合のみ。）主な違いを説明して、ユーザーが選べるよう支援してください。通常の Consumer API（KafkaConsumer、コンシューマーグループ）では、各パーティションは必ず 1 つのコンシューマーに割り当てられるため、並列処理の上限はパーティション数で、キー単位の順序が維持されます。Share Consumer API（KafkaShareConsumer、シェアグループ、「Kafka のキュー」）では、複数のシェアコンシューマーが同じパーティションからメッセージを消費できます。そのため、複数のコンシューマーが協調してメッセージを同時に処理する、キューに似たメッセージングセマンティクスが得られ、コンシューマー数をパーティション数を超えて増やせます（トレードオフとして、キー単位の順序は保証されず、各レコードの確認応答が必要です）。通常の Consumer API をデフォルトにします。キュー／ワークキューのセマンティクス、競合するコンシューマー、協調的な同時処理、またはパーティション数とは独立したコンシューマーのスケーリングをユーザーが求める場合は、Share Consumer API を選択します。ユーザーが Share Consumer API を選んだ場合は、生成する前に references/share-consumer.md を読みます。この API には Kafka 4.x クライアントが必要です（リファレンスのビルドファイルでは、すでに kafka.version=4.2.1／confluent.version=8.2.1 に固定されています）。また、クラスター側での機能有効化も必要です。 Schema Registry を使用するかどうかは尋ねず、必ず含めてください。デフォルトは Avro です（JVM では標準的です。生成された SpecificRecord クラスにより、コンパイル時の型安全性とコンパクトなワイヤーフォーマットが得られます）。ユーザーが希望する場合は JSON Schema または Protobuf を提案してください。希望を示唆している場合に限り尋ね、それ以外は Avro を使用します。ターゲットが WarpStream で、ユーザーが WarpStream 組み込みのスキーマレジストリを使用している場合は、Avro と Protobuf しかサポートされないこと（GET /schemas/types は ["AVRO","PROTOBUF"] を返します）を伝えてください。Avro はすでにデフォルトなので、Protobuf を希望する場合を除き変更は不要です。

エージェントによくある誤り

考え	実際
「ユーザーは Spring Boot と言ったので、新しい Maven プロジェクトをひな形から作ろう」	既存のアプリがある場合は、その場で変更します。ビルドファイルに依存関係を追加し、シリアライザーを統合してください。競合する別のプロジェクトをひな形から作成しないでください。
「Python スキルのように JSON Schema をデフォルトにしよう」	このスキルでは、JVM 上のデフォルトは Avro です。生成された SpecificRecord クラスは標準的で、型安全です。JSON Schema／Protobuf を使用するのは、ユーザーが希望する場合に限ります。
「auto.register.schemas はデフォルトのままにしよう」	シリアライザーのデフォルトは true です。必ず false に設定し、CachedSchemaRegistryClient.register(...) で明示的に登録してください。暗黙の自動登録は、本番環境では危険です。
「送信メソッド内で KafkaProducer を作成しよう」	main() で KafkaProducer インスタンスを 1 つ作成し、パラメーターとして渡します。プロデューサーはスレッドセーフで、作成コストも高いためです。
「Python の同期プロデューサーのように、レコードヘッダーにスキーマ ID を書き込もう」	Java Confluent シリアライザーは、スキーマ ID をワイヤーフォーマット（マジックバイト + 4 バイトの ID）に自動的に埋め込みます。ヘッダーを独自に作成しないでください。
「Avro にはクラスパス上の .avsc があればよい」	Avro SpecificRecord クラスは、ビルド時に avro-maven-plugin（Maven）または com.github.davidmc24.gradle.plugin.avro プラグイン（Gradle）が src/main/avro/*.avsc から生成します。プロデューサーは生成されたクラス（例：com.example.kafka.Transaction）をインポートします。参照する前に mvn generate-sources ／ ./gradlew generateAvroJava を実行してください。
「WakeupException は一般的に捕捉して無視しよう」	WakeupException は、シャットダウンフック内で consumer.wakeup() を呼び出した際に発生する想定内のシグナルです。ポーリングループを抜けるため、限定的に捕捉し、その後 finally で close() を呼び出してください。
「Consumer といえば KafkaConsumer だから、Share Consumer API について尋ねる必要はない」	コンシューマーを求められた場合は、必ず Consumer API と Share Consumer API のどちらかを尋ねてください（質問 #10）。デフォルトは通常の Consumer API とし、キュー形式／協調型のコンシュームには Share Consumer API（KafkaShareConsumer）を提案してください。
「Confluent Maven リポジトリは忘れてもよい。シリアライザーは Maven Central にある」	io.confluent:kafka-avro-serializer などは https://packages.confluent.io/maven/ から解決されます。ビルドファイルで、このリポジトリを必ず宣言してください。
「schema.registry.url は KafkaProducer の設定にだけ指定すればよい」	シリアライザーは、同じプロデューサー／コンシューマーの Properties から schema.registry.url（および basic.auth.*）を読み取ります。シリアライザーが参照できるよう、クライアントのプロパティに SR 設定を指定してください。

ステップ 1b：理解内容の確認

すべての回答を収集したら、コードを生成する前に確認用の要約を提示してください。

Before I generate the project, let me confirm:
- Project type: [Greenfield scaffold / Migration of existing code]
- Environment: [Confluent Cloud (SASL_SSL) / Local Docker (PLAINTEXT) / WarpStream]
- Schema format: [Avro (default) / JSON Schema / Protobuf]
- Build tool: [Maven / Gradle]
- Components: [Producer only / Consumer only / Both]
- Send pattern: [Async callback / Synchronous] (if producer)
- Consumer API: [Consumer API / Share Consumer API] (if consumer)
- Schema: [brief description of user's data fields]
- Topic: [topic name]
- Consumer group: [group ID] (if consumer)

Does this look right?

ステップ 2 に進む前に、ユーザーの確認を待ってください。ユーザーが何かを訂正した場合は、理解内容を更新して再確認してください。

ステップ 2：プロジェクトの生成

判断フローチャート

digraph decisions {
  "Q1: Existing app?" -> "Migration path:\nmodify existing code +\nadd deps to build file" [label="yes"];
  "Q1: Existing app?" -> "Q2: Environment?" [label="no / greenfield"];
  "Q2: Environment?" -> "Cloud config\n(SASL_SSL)" [label="Confluent Cloud"];
  "Q2: Environment?" -> "Local Docker config\n(PLAINTEXT) + docker-compose.yml" [label="local / docker / OSS"];
  "Q2: Environment?" -> "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" [label="WarpStream"];
  "Cloud config\n(SASL_SSL)" -> "Q4: Send pattern?";
  "Local Docker config\n(PLAINTEXT) + docker-compose.yml" -> "Q4: Send pattern?";
  "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" -> "Q4: Send pattern?";
  "Q4: Send pattern?" -> "AvroProducer.java\n(send + Callback)" [label="async callback"];
  "Q4: Send pattern?" -> "AvroProducerSync.java\n(send().get())" [label="synchronous"];
}

ユーザーが選択したディレクトリに、次のファイル構成を作成してください（Maven の構成を示しています。Gradle の場合、ビルドファイルのみが異なり、ソースツリーは同じです）。

<project-dir>/
├── pom.xml                       # Maven (or build.gradle for Gradle)
├── docker-compose.yml            # (local Docker path only)
├── .properties.example           # template for credentials
├── README.md
└── src/
    ├── main/
    │   ├── java/com/example/kafka/
    │   │   ├── KafkaConfig.java   # shared config loading + connectivity verification
    │   │   ├── AvroProducer.java  # (if producer requested; or AvroProducerSync.java)
    │   │   └── AvroConsumer.java  # (if consumer requested; or AvroShareConsumer.java for the Share Consumer API)
    │   └── avro/
    │       └── value.avsc         # Avro schema (codegen source; value.schema.json / value.proto for other formats)
    └── test/
        └── java/com/example/kafka/
            └── AppTest.java       # unit tests (always generated)

セキュリティ

.properties ファイルは、API キーやシークレットを含むため、絶対に読み取ったり、開いたり、表示したりしないでください。プレースホルダー値を含む .properties.example のみを生成してください。接続問題のデバッグを求められた場合は、.properties の値をユーザー自身で確認するよう依頼し、ファイルを読み取らないでください。.gitignore に .properties を追加してください。

基本原則

これらの原則は、Kafka Java クライアントで本番環境において最もよく発生する問題を防ぐために重要です。

1. プロデューサー／コンシューマーのインスタンスを再利用します。 メッセージごとに KafkaProducer を作成するとコストがかかります。インスタンスごとに TCP 接続を開き、SASL ハンドシェイクを実行し、メタデータを取得するためです。KafkaProducer はスレッドセーフなので、1つ作成して共有してください。送信ヘルパーはプロデューサーをパラメーターとして受け取り、インスタンスを作成することはありません。

2. 必ず Schema Registry を使用し、デフォルトでは Avro を使用します。 Schema Registry はプロデューサーとコンシューマー間の契約を強制します。これがないと、スキーマの変更によってコンシューマーが気付かないうちに破損します。このスキルではデフォルトで Avro を使用します。生成された SpecificRecord クラスにより、JVM でコンパイル時の型安全性が確保され、コンパクトなワイヤーフォーマットが得られるためです。JSON Schema と Protobuf も代替形式としてサポートされています。

   メッセージを生成する前に、スキーマを個別の明示的な手順として登録してください。専用の registerSchema() メソッドを使用して schemaRegistryClient.register(subject, parsedSchema) を呼び出し、エラー（認証エラー、ネットワークエラー、権限拒否）をすぐに伝播させてください。決して握りつぶしてはいけません。その後、シリアライザーを auto.register.schemas=false および use.latest.version=true に設定します。これにより、シリアライザーによる自動登録が暗黙的に行われなくなり、アプリケーションの起動時ではなく CI/CD でスキーマを登録する本番環境の運用に沿った構成になります。

   選択した形式に対応するシリアライザーを使用してください。

   • Avro（デフォルト）： io.confluent.kafka.serializers.KafkaAvroSerializer ／ KafkaAvroDeserializer。コンシューマーで specific.avro.reader=true を設定し、生成された SpecificRecord クラスにデシリアライズします。
   • JSON Schema： io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer ／ KafkaJsonSchemaDeserializer。コンシューマーで POJO クラスを json.value.type に設定します。references/JsonSchemaProducer.java と references/JsonSchemaConsumer.java を参照してください。
   • Protobuf： io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer ／ KafkaProtobufDeserializer。protobuf-maven-plugin を使って .proto からクラスを生成し、コンシューマーで specific.protobuf.value.type を設定します。

3. 送信パターンは意図を持って選択します。 KafkaProducer.send() は Future<RecordMetadata> を返し、内部では常に非同期で動作します。

   • コールバック付き非同期送信（references/AvroProducer.java）：producer.send(record, (metadata, exception) -> {...})。ノンブロッキングで、スループットが最も高くなります。デフォルトとして推奨します。
   • 同期送信（references/AvroProducerSync.java）：producer.send(record).get()。レコードごとにブロックし、確認の確実性が最も高い一方で、スループットは最も低くなります。バッチ処理／ETL に最適です。 どちらの場合も、コンシューマーは単一スレッドの poll() ループを使用します。

4. 正常にシャットダウンします。 プロデューサーは、バッファー内のレコードが失われないよう、flush() の後に（finally ブロックで）close() する必要があります。コンシューマーではウェイクアップパターンを使用します。Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup)) を登録し、consumer.wakeup() が WakeupException を発生させるまでポーリングループを実行します。例外は範囲を絞って捕捉してループを終了し、finally で close() します。close() によりオフセットがコミットされ、コンシューマーグループから正常に離脱できるため、不要なリバランスを回避できます。

5. Confluent Cloud、ローカル Docker、WarpStream をサポートします。 Confluent Cloud では、sasl.jaas.config を使って PLAIN メカニズムの SASL_SSL を構成し、API キーを .properties から読み込みます。ローカル Docker では、認証なしで PLAINTEXT を使用します。WarpStream では、ユーザーのデプロイ環境に応じて SASL_SSL または PLAINTEXT を使用し、references/warpstream-optimization.md にある Java のオーバーライドを適用します。KAFKA_ENV 設定（cloud、local、warpstream）によって、KafkaConfig が使用する接続方法が決まります。すべての設定は java.util.Properties を使って .properties ファイルから読み込み、ファイルに存在しないキーについてはプロセスの環境変数にフォールバックします。

6. 実行前に接続を確認します。 AdminClient.listTopics() を使用して、ブローカーに接続できることとトピックが存在することを確認してから、メッセージの生成または消費を行います。/subjects に対する HTTP ヘルスチェックで、Schema Registry への接続も確認します。

7. ドメインイベントには必ずメッセージキーを設定します。 エンティティまたはイベントストリームを表すメッセージ（注文イベント、ユーザー操作、デバイスのテレメトリー、トランザクション）には、エンティティ識別子を ProducerRecord のキーとして渡します。Kafka はキーに基づいてパーティション分割するため、同じキーのメッセージは同じパーティションに送られ、順序が保たれます。これは OrderCreated → OrderUpdated → OrderCancelled のようなストリームでは特に重要です。どのフィールドがエンティティを識別するかをユーザーに確認し、それをキーとして使用します。順序が重要でないとユーザーが明示した場合に限り、キーに null を使用します。

   WarpStream の例外： WarpStream では、キーに null を指定するとスティッキーパーティショニングが有効になります。これにより、より大きなバッチが作成され、スループットとコストが大幅に改善します。ユースケースでエンティティごとの順序が不要な場合（独立したテレメトリー、ステートレスなメトリクス、ログ）は、null キーを推奨し、スループットが向上する理由を説明します。エンティティごとの順序が必要な場合は、引き続きキーを設定します。正確性を優先してください。イベントでキーごとの順序が必要かどうかをユーザーに確認して判断します。

KafkaConfig.java

このクラスは、設定の読み込み（java.util.Properties を使用して .properties ファイルを読み込み、プロセス環境変数をフォールバックとして使用）と接続の確認を処理します。テンプレートとして references/KafkaConfig.java を使用します。クライアントの基本 Properties（KAFKA_ENV に応じた SASL_SSL または PLAINTEXT）、シリアライザーが読み取る Schema Registry の設定キー、および verifyKafkaSetup／verifySchemaRegistry ヘルパーを作成します。

Producer Patterns

ユーザーが非同期コールバックを選択した場合は、references/AvroProducer.java を使用します。同期を選択した場合は、references/AvroProducerSync.java を使用します。JSON Schema には、references/JsonSchemaProducer.java を使用します。すべてに共通する重要な点は次のとおりです。

• 送信ヘルパーは KafkaProducer をパラメーターとして受け取り、自身で作成することはありません。プロデューサーは main() で一度だけ作成します。
• registerSchema() は CachedSchemaRegistryClient を使用して明示的に登録し、スキーマ ID を返します。エラーは呼び出し元に伝播します。シリアライザーは auto.register.schemas=false、use.latest.version=true に設定します。
• スキーマ ID は Confluent ワイヤーフォーマットに自動的に埋め込まれます。ヘッダーとして追加しないでください。
• finally ブロックで flush() の後に close() を呼び出します。

コンシューマーパターン

通常のコンシューマー API（デフォルト）。 references/consumer.md を読み、references/AvroConsumer.java（JSON Schema の場合は references/JsonSchemaConsumer.java）を使用します。主なポイント：

• Schema Registry のデシリアライザーを使ってデシリアライズします。生の String／JSON へのフォールバックはありません。
• Avro：specific.avro.reader=true を設定し、生成された SpecificRecord 型に読み込みます。
• AK 4.0 以降では、次世代のリバランスプロトコル（KIP-848）を使用するために group.protocol=consumer を設定します。これにより、ワールドストップ型のリバランスがなくなります。references/consumer.md を参照してください。
• enable.auto.commit を無効にし、処理後に明示的にコミットします。commitSync と commitAsync のどちらを使うかは、references/consumer.md に従って検討します。
• Wakeup を使ったグレースフルシャットダウン：シャットダウンフックで consumer.wakeup() を呼び出し、ループでは WakeupException を捕捉し、finally で close() を呼び出します。
• シャットダウンするまで、poll(Duration.ofMillis(1000)) を継続的に呼び出すループを実行します。

Share Consumer API（「Queues for Kafka」）。 質問 #10 でユーザーが Share Consumer API を選択した場合は、references/share-consumer.md を読み、AvroConsumer.java の代わりに references/AvroShareConsumer.java を使用します。この API は KafkaShareConsumer を使用するため、同じシェアグループ内の複数のインスタンスが、キューのようなセマンティクスで同じパーティションから協調して消費します。上記と同じ Schema Registry デシリアライザーと Wakeup によるシャットダウンに加え、レコードごとの acknowledge(record, AcknowledgeType.ACCEPT/RELEASE/REJECT) と、クラスタ側での機能有効化が必要です（ローカルでは share.version=1、Confluent Cloud では「Queues for Kafka」を有効にします）。参照用のビルドファイルでは Kafka 4.x クライアントがすでに固定されているため、バージョン変更は不要です。そのまま使用してください。機能要件をユーザーに伝えてください。

スキーマ

ユーザーのデータドメインに合ったスキーマを生成します。

Avro（デフォルト）： スキーマを src/main/avro/value.avsc に配置し、ビルドプラグインで SpecificRecord クラスを生成します。references/value.avsc を出発点として、ユーザーのドメインに合わせて調整します。レコードの name が生成される Java クラス名になり、namespace がそのパッケージになります（例：name: "Transaction"、namespace: "com.example.kafka" → com.example.kafka.Transaction）。references/schema-generation-rules.md に厳密に従ってください。

JSON Schema： src/main/resources/value.schema.json に配置し、POJO と組み合わせます。Protobuf： .proto ファイルを src/main/proto/ の下に配置します。

複数イベントのトピック（上級）

ユーザーが単一のトピックで複数のイベントタイプを扱うことを説明した場合は、references/multi-event-guide.md に従ってください。ユーザーが単一のトピック上の複数のイベントタイプを明示的に説明した場合に限り、複数イベント／ユニオンスキーマを提案します。

docker-compose.yml（ローカル Docker の場合のみ）

ユーザーがローカル Docker を選択した場合、references/docker-compose.yml をテンプレートとして docker-compose.yml を必ず生成します。単一ノードの Kafka ブローカー（KRaft モードの confluentinc/confluent-local）と Confluent Schema Registry を起動します。ユーザーは docker compose up -d を実行します。

重要： confluentinc/confluent-local に組み込まれているリスナー名は PLAINTEXT（29092）、PLAINTEXT_HOST（9092）、CONTROLLER（29093）です。独自のリスナー名を作成しないでください。イメージ内部の設定と競合し、起動ループが発生します。KAFKA_ADVERTISED_LISTENERS と KAFKA_LISTENERS を上書きする場合は、これらの名前をそのまま使用してください。内部の PLAINTEXT リスナーでは、Docker ネットワーク内で Schema Registry がブローカーに接続できるよう、kafka ホスト名をアドバタイズする必要があります。

.properties.example

対象環境に適した .properties.example を生成します。これは標準的な Java プロパティファイル（key=value）で、KafkaConfig が java.util.Properties を介して読み込みます。

Confluent Cloud：

KAFKA_ENV=cloud
BOOTSTRAP_SERVER=pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
API_KEY=your-api-key
API_SECRET=your-api-secret
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=https://psrc-xxxxx.us-east-2.aws.confluent.cloud
SR_API_KEY=your-sr-api-key
SR_API_SECRET=your-sr-api-secret
CLIENT_ID=java-client
GROUP_ID=java-consumer-group

ローカル Docker：

KAFKA_ENV=local
BOOTSTRAP_SERVER=localhost:9092
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=http://localhost:8081
CLIENT_ID=java-client
GROUP_ID=java-consumer-group

WarpStream：

KAFKA_ENV=warpstream
BOOTSTRAP_SERVER=your-warpstream-bootstrap-url:9092
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=http://your-schema-registry:8081
CLIENT_ID=java-client,ws_az=us-east-1a
GROUP_ID=java-consumer-group
# If your WarpStream deployment requires SASL auth, uncomment:
# API_KEY=your-api-key
# API_SECRET=your-api-secret
# If using Confluent Cloud Schema Registry, uncomment:
# SR_API_KEY=your-sr-api-key
# SR_API_SECRET=your-sr-api-secret

ビルドファイル

Maven（デフォルト）： references/pom.xml を使用します。Confluent Maven リポジトリ、kafka-clients、Avro シリアライザー + avro + avro-maven-plugin（src/main/avro からコードを生成）、ロギングバインディング、テスト用の JUnit 5 + Mockito が宣言されています。設定の読み込みには JDK 組み込みの java.util.Properties を使用し、追加の依存関係は必要ありません。JSON Schema または Protobuf シリアライザーへの切り替え方法はコメントに記載されています。

Gradle： references/build.gradle を使用します。同じ依存関係に加えて、コード生成用の com.github.davidmc24.gradle.plugin.avro プラグインを使用します。

生成されたコードで使用するすべての依存関係をビルドファイルに記載する必要があります。ユーザーが mvn package / ./gradlew build を実行し、未解決のシンボルや ClassNotFoundException エラーなしで実行できるようにしてください。

README.md

references/readme-template.md に沿って README を生成します。実際に生成された内容に合わせて調整してください。コンシューマーのみを要求された場合はプロデューサーのセクションを省略し、Confluent Cloud プロジェクトでは Docker のセクションを省略し、適切なビルドツールのコマンドを使用します。

テスト

必ず src/test/java/com/example/kafka/AppTest.java に単体テストを生成してください。テンプレートとして references/AppTest.java を使用します。テストは、稼働中の Kafka クラスターや Schema Registry がなくても実行できるようにしてください。kafka-clients の MockProducer / MockConsumer と、Schema Registry クライアント用の Mockito を使用します。

テストでは、次のプロパティを検証します。

1. KafkaConfig：KAFKA_ENV=cloud の場合、baseProperties() が SASL_SSL + PLAIN（sasl.jaas.config を設定）を生成し、KAFKA_ENV=local の場合は SASL を使用しない PLAINTEXT を生成します。verifyKafkaSetup() がモックした AdminClient に対して適切な boolean を返すことも検証します。
2. プロデューサー（生成されている場合）：送信ヘルパーがプロデューサーをパラメーターとして受け取り（決して生成しない）、クラス内でインスタンス化される KafkaProducer がちょうど 1 つであることを検証します。また、MockProducer を使ってバッチを送信し、メッセージごとに期待されるキーを含む履歴エントリーが 1 つ記録されることを検証します。同期処理の場合は、送信結果が解決される（.get()）ことを検証します。
3. コンシューマー（生成されている場合）：値のデシリアライザーに raw の StringDeserializer ではなく Schema Registry デシリアライザー（KafkaAvroDeserializer / KafkaJsonSchemaDeserializer / KafkaProtobufDeserializer）を使用し、wakeup + close() のシャットダウンパターンを使用することを検証します。レコードを投入した MockConsumer から、期待されるデシリアライズ済みの値が得られることを検証します。
4. スキーマファイル：Avro の value.avsc が、トップレベルの doc とすべてのフィールドの doc を持つ record として解析されることを検証します（references/schema-generation-rules.md を参照）。JSON Schema では、type: object、title、各フィールドの説明、およびタイムスタンプフィールドの format: date-time を確認します。

すべてのファイルを生成した後、mvn test（または ./gradlew test）を実行してテストが成功することを確認してください。テストが失敗した場合は、成功するまでテストではなく生成されたコードを修正してください。

ステップ 3：ユーザーを案内する

ファイルを生成したら、対象環境に応じた手順を案内します。生成された内容に合わせて調整してください。

Confluent Cloud：

1. .properties.example を .properties にコピーし、Confluent Cloud の認証情報（ブートストラップサーバー、API キー、Schema Registry の URL）を入力します。これらはすべて、Confluent Cloud Console のクラスター／環境の設定で確認できます。
2. Avro クラスを生成してビルドします：mvn generate-sources && mvn package（または ./gradlew build）。
3. プロデューサーが生成された場合、初回実行時に registerSchema() を通じてスキーマが明示的に登録されます（auto.register.schemas=false）。コンシューマーのみが生成された場合は、Console の Schema Registry で、対象トピックの value サブジェクトにスキーマを手動で登録します。
4. プロデューサーを実行します：mvn exec:java -Dexec.mainClass=com.example.kafka.AvroProducer（または ./gradlew run）。
5. コンシューマーを実行します：mvn exec:java -Dexec.mainClass=com.example.kafka.AvroConsumer。

ローカル Docker：

1. Kafka と Schema Registry を起動します：docker compose up -d。
2. .properties.example を .properties にコピーします（ローカル Docker 用のデフォルト値は入力済みです）。
3. ビルドします：mvn package（または ./gradlew build）。
4. 自動作成が無効になっている場合は、トピックを作成します：docker compose exec kafka kafka-topics --create --topic demo-topic --bootstrap-server localhost:29092。
5. 上記の手順でプロデューサー／コンシューマーを実行します。
6. 終了したら、docker compose down を実行します（保存データを削除するには -v を追加します）。

WarpStream：

1. .properties.example を .properties にコピーし、WarpStream のブートストラップサーバー、Schema Registry の URL、および（該当する場合は）認証情報を入力します。ゾーンを考慮したルーティングを行うには、CLIENT_ID に ws_az=<availability-zone> を含めます。
2. ビルドします：mvn package（または ./gradlew build）。
3. トピックが存在しない場合は作成します。
4. 上記の手順でプロデューサー／コンシューマーを実行します。

Share Consumer が生成された場合（AvroShareConsumer.java）：実行前に、クラスターで share 機能を有効にし、share グループの開始位置を設定する必要があります（references/share-consumer.md を参照）。ローカル環境では、docker compose exec kafka kafka-features --bootstrap-server localhost:9092 upgrade --feature share.version=1 を実行し、次に kafka-configs --bootstrap-server localhost:9092 --alter --group <group> --add-config share.auto.offset.reset=earliest を実行します。Confluent Cloud では、（Dedicated）クラスターで「Queues for Kafka」が有効になっていることを確認します。share コンシューマーを mvn exec:java -Dexec.mainClass=com.example.kafka.AvroShareConsumer で実行し、協調的なコンシュームを確認するには、同じ GROUP_ID を指定した複数のインスタンスを起動します。

WarpStream のユーザーには、プロデュースのレイテンシー（p50 で約 250ms）は標準の Kafka より高く、これは想定どおりであることを伝えてください。スループットが低い場合は、references/warpstream-optimization.md のオーバーライドが適用されていることを確認してください（特に enable.idempotence=false と、大きなバッチ／フェッチサイズ）。