developing-kafka-python-client

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

まず、「Confluent Kafka Python Client スキルを使用して、このプロジェクトを進めます。」と宣言してください。

Confluent Kafka Python Client の作成

confluent-kafka-python を使用して、Kafka にメッセージを送信する、および／または Kafka からメッセージを受信する、本番環境対応の Python プロジェクトを生成します。対象環境は Confluent Cloud（マネージド）、Local Docker（オープンソースの Kafka）、WarpStream（オブジェクトストレージをバックエンドとする Kafka 互換サービス）の 3 種類に対応しています。また、プロデューサーのスタイルは AsyncIO（ノンブロッキング）と Synchronous（ブロッキング）の 2 種類です。生成されるコードは Confluent のベストプラクティスに従います。

ステップ 1：要件の確認

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

必須の確認ゲート — ユーザーがすべての質問に回答していても、省略しないでください。 ファイルを書き込む前に、必ず以下の内容を含むメッセージを 1 回送信してください。

1. 抽出した回答を、短い箇条書きで要約する（例：「対象：Confluent Cloud · コンポーネント：プロデューサー + コンシューマー · プロデューサーのスタイル：非同期 · 新規作成：はい」）。
2. 未回答の質問があれば、同じメッセージ内で追加質問する。
3. 作業を進める前に、内容の確認または訂正をユーザーに明示的に求める。

その後、ユーザーの返信を待ってください。要約と同じターンでファイルを生成してはいけません。また、必要事項がすべて指定されたプロンプトは、そのまま生成への同意を意味するとみなして先に進んではいけません。プロンプトの解釈に誤りがないかを確認するため、この要約は、質問 #1～#8 にすべて事前回答がある場合でも必須です。このゲートを省略できるのは、この会話内ですでに要約を提示し、ユーザーから確認を受けている場合に限ります。

#1、#2、#3 のいずれかがプロンプトで回答されていない場合、デフォルトを推測せず、必ず質問してください。

1. 既存のアプリケーションに Kafka を追加しますか、それとも新しく始めますか？
   • ユーザーが既存の Python コードに言及している場合（既存のプロジェクトや main.py がある、Flask／FastAPI／Django を使用しているなど）は、新しいプロジェクトをスキャフォールディングしないでください。代わりに、（a）既存のプロデューサーまたはデータ送信コードを特定し、（b）Schema Registry に登録済みのスキーマがあるかを確認し、（c）リファレンスファイルのパターンに従って、既存のコードに Schema Registry 統合を追加してください。生成するのは不足しているファイル（例：common.py、schemas/value.schema.json）のみにして、既存のコードは直接変更してください。
   • ユーザーが Schema Registry を使わずに Kafka にすでにデータを送信している場合（スキーマなしの場合）は、移行を支援してください。（1）既存のメッセージ構造から JSON Schema を生成し、（2）それを登録し、（3）既存の生の producer.produce() 呼び出しを、シリアライザーを使用する呼び出しに置き換えてください。既存のコードは破棄しないでください。
   • 新しく始める場合は、以下の完全なスキャフォールドに進んでください。
2. 対象環境は何ですか？ — Confluent Cloud、ローカル Kafka（Docker）、または WarpStream。ユーザーが言及していない場合でも、必ず確認してください。「オープンソース」「ローカル」「docker」「セルフホスト可能」と言及している場合、またはクラウドアカウントなしで Kafka を試したいだけの場合は、ローカル Dockerを選んでください。「Confluent Cloud」「CC」と言及している場合、または既存のクラウド認証情報がある場合は、Confluent Cloudを選んでください。「WarpStream」と言及している場合は、WarpStreamを選んでください。希望がないと確認できた場合は Confluent Cloud をデフォルトにしますが、必ず最初に確認してください。
   • WarpStream の場合： references/warpstream-optimization.md を読み、そこに記載された librdkafka のオーバーライドを適用してください。主な変更点：冪等性を無効にし、バッチサイズとインフライトリクエスト数を大幅に増やし、フェッチサイズを大きく設定し、ゾーンを考慮したルーティングのために client.id に ws_az=<az> を追加します。エンティティ単位の順序保証が必要な場合を除き、スティッキーパーティショニングには null のメッセージキーを優先してください。
3. プロデューサー、コンシューマー、または両方ですか？
4. プロデューサーは非同期ですか、同期ですか？（プロデューサーを使用する場合のみ。）ユーザーが選択できるよう支援してください。
   • AsyncIO Producer（AIOProducer）：イベントループ上で実行され、ブロックしてはならないコード（FastAPI／Starlette、aiohttp、Sanic、asyncio ワーカーなど）で使用します。
   • Synchronous Producer（Producer）：スクリプト、バッチジョブ、スレッドやプロセスをユーザーが制御し、poll()／flush() を直接呼び出せる最高スループットのパイプラインで使用します。 ユーザーが非同期フレームワーク（FastAPI、aiohttp、Sanic）に言及している場合、または asyncio を使用している場合は、AsyncIOをデフォルトにします。スクリプト、バッチ、ETL に言及している場合、または希望がない場合は、Synchronousをデフォルトにします。
5. 使用したい既存のスキーマはありますか？ ある場合は、貼り付けるかファイルパスを知らせるようユーザーに依頼し、生成する代わりに schemas/value.schema.json として使用してください。ない場合は、データのフィールドについて確認してください。
6. どのようなデータを生成しますか？（既存のスキーマがない場合のみ。適切な JSON Schema とサンプルデータを生成できるよう、フィールド名と型を確認してください。）
7. トピック名は何ですか？（デフォルト：demo-topic）
8. コンシューマーグループ ID は何ですか？（コンシューマーを使用する場合のみ。デフォルト：python-consumer-group）

Schema Registry については質問しないでください。常に含めてください。Confluent Cloud とローカル Docker では、常に JSON Schema を使用してください。対象が WarpStream の場合は、使用する Schema Registry の実装を確認してください。WarpStream 組み込みのスキーマレジストリは Avro と Protobuf のみをサポートするため（GET /schemas/types は ["AVRO","PROTOBUF"] を返します）、それを使用している場合は、Avro と Protobuf のどちらを希望するか確認してください（デフォルトは Avro）。別の SR（例：Confluent Cloud Schema Registry）を使用している場合は、JSON Schema で問題ありません。

エージェントがよく陥る間違い

考え	実際
「ユーザーが FastAPI に言及したから、非同期だとわかっている。質問は省略しよう」	それでも確認してください。FastAPI と並行して、同期のバックグラウンドワーカーを使いたい場合もあります。
「より広く使われている Avro を使おう」	このスキルでは、デフォルトで JSON Schema を使います。例外： WarpStream の組み込みスキーマレジストリは Avro と Protobuf のみをサポートしています。ユーザーが WarpStream SR を使用している場合は、デフォルトで Avro を使ってください。別の SR（例：Confluent Cloud SR）を使用している場合は、Kafka 環境にかかわらず JSON Schema を使用できます。
「シンプルにするため、スキーマレジストリは省略しよう」	スキーマレジストリは必須です。すべてのプロジェクトに含めます。
「利便性のために auto.register.schemas=True を使おう」	常に False にしてください。明示的な登録は基本原則です。
「produce() 内でプロデューサーを作成しよう。そのほうがすっきりする」	main() でプロデューサーインスタンスを1つ作成し、パラメーターとして渡してください。必ずそうします。
「ユーザーが同期処理を希望しているから、コンシューマーも同期にしよう」	コンシューマーは常に非同期（AIOConsumer）です。これは意図的な設計上の判断です。
「スキーマ ID 用に AIOProducer に headers= を追加しよう」	AIOProducer.produce() はヘッダーを指定すると NotImplementedError を送出します。ヘッダーを使うのは同期プロデューサーのみです。
「AsyncJSONSerializer を AsyncAvroSerializer に置き換えて、呼び出し側はそのままにしよう」	JSONSerializer は最初の引数に schema_str を取り、AvroSerializer は最初の引数に schema_registry_client を取ります。フォーマットをまたいで位置引数で呼び出すと、TypeError: ... got multiple values for argument 'schema_registry_client' が送出されます。両方とも常にキーワード引数で渡してください。
「プロデューサー設定に message.max.bytes=64000000 を、コンシューマーに fetch.max.bytes=50242880 を設定しよう。互いに独立しているから」	librdkafka ではそうなりません。Java のロールごとの max.request.size とは異なり、message.max.bytes はクライアント全体の設定です。そのため、get_kafka_config() をプロデューサーとコンシューマーで共有すると、コンシューマーにも引き継がれます。librdkafka はコンシューマーの構築時に fetch.max.bytes >= message.max.bytes を強制し、KafkaError{_INVALID_ARG, ... "fetch.max.bytes must be >= message.max.bytes"} を送出します。この制約を満たすように設定された WarpStream の librdkafka コンシューマー値を references/warpstream-optimization.md（fetch.max.bytes=67108864）で使用してください。

Step 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: [JSON Schema / Avro / Protobuf] (Avro or Protobuf if using WarpStream's built-in SR)
- Components: [Producer only / Consumer only / Both]
- Producer style: [AsyncIO (AIOProducer) / Synchronous (Producer)] (if applicable)
- Schema: [brief description of user's data fields]
- Topic: [topic name]
- Consumer group: [group ID] (if consumer)

Does this look right?

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

Step 2: プロジェクトの生成

判断フローチャート

digraph decisions {
  "Q1: Existing app?" -> "Migration path:\nmodify existing code" [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)" -> "Q3: Components?";
  "Local Docker config\n(PLAINTEXT) + docker-compose.yml" -> "Q3: Components?";
  "WarpStream config\n(apply overrides from\nreferences/warpstream-optimization.md)" -> "Which SR?\n(WarpStream SR → Avro/Protobuf;\nother SR → JSON Schema)";
  "Which SR?\n(WarpStream SR → Avro/Protobuf;\nother SR → JSON Schema)" -> "Q3: Components?";
  "Q3: Components?" -> "Q4: Async or sync?" [label="producer requested"];
  "Q3: Components?" -> "Generate consumer\n(always async AIOConsumer)" [label="consumer only"];
  "Q4: Async or sync?" -> "AIOProducer path\nAsyncJSONSerializer\n(no headers support)" [label="async / event-loop"];
  "Q4: Async or sync?" -> "Producer path\nJSONSerializer\n(header-based schema ID)" [label="sync / batch / ETL"];
}

ユーザーが選択したディレクトリに、次のファイル構成を作成してください。

<project-dir>/
├── producer.py          # (if requested)
├── consumer.py          # (if requested)
├── common.py            # shared config loading + verification helpers
├── schemas/
│   └── value.schema.json # JSON Schema (or value.avsc / value.proto when using WarpStream SR)
├── tests/
│   └── test_project.py  # unit tests (always generated)
├── .env.example         # template for credentials
├── requirements.txt
├── docker-compose.yml   # (local Docker path only)

セキュリティ

.env ファイルは絶対に読み込んだり、開いたり、表示したりしないでください。API キーやシークレットが含まれています。プレースホルダー値を使った .env.example のみを生成してください。ユーザーから接続の問題のデバッグを依頼された場合は、自分で .env の値を確認するようユーザーに依頼してください。ファイルを読んではいけません。

基本原則

これらの原則が重要なのは、Kafka の Python クライアントで最も一般的な本番環境の問題を防げるためです。

1. プロデューサーインスタンスを再利用します。 メッセージごとに新しいプロデューサーを作成するとコストがかかります。インスタンスごとに新しい TCP 接続を開き、SASL ハンドシェイクを行い、メタデータを取得するためです。プロデューサーを1つ作成し、すべてのメッセージで再利用してください。produce 関数はプロデューサーをインスタンス化するのではなく、パラメーターとして受け取るようにしてください。

2. JSON Schema を使って、必ずスキーマレジストリを使用します。 スキーマレジストリは、プロデューサーとコンシューマーの間の契約を強制します。これがなければ、スキーマの変更によって後続のコンシューマーが予期せず動作しなくなります。このスキルでは、デフォルトで JSON Schema を使用します。スキーマレジストリは Avro、Protobuf、JSON Schema をサポートしています。JSON Schema を選ぶ理由は、（1）Python にはコード生成の手順を必要としない、組み込みの JSON サポートがあること、（2）confluent-kafka-python が JSONSerializer／JSONDeserializer をすぐに使える形で提供していること、（3）すでに JSON／dict データを扱っている Python 開発者にとって最も取り組みやすい形式であることです。 WarpStream スキーマレジストリの例外： WarpStream に組み込まれているスキーマレジストリは Avro と Protobuf のみをサポートしており、その GET /schemas/types エンドポイントは ["AVRO","PROTOBUF"] を返します。WarpStream の SR を使用する場合、JSON Schema は利用できません。ユーザーが WarpStream を実行し、組み込みの SR を使用している場合は、デフォルトで Avro を使用してください（ユーザーの希望に応じて Protobuf を使用してもかまいません）。confluent_kafka.schema_registry.avro の AvroSerializer／AvroDeserializer（または confluent_kafka.schema_registry.protobuf の ProtobufSerializer／ProtobufDeserializer）を使用してください。非同期プロデューサーでは、confluent_kafka.schema_registry._async.avro の AsyncAvroSerializer（または confluent_kafka.schema_registry._async.protobuf の AsyncProtobufSerializer）を使用してください。schemas/value.schema.json ではなく、schemas/value.avsc（Protobuf の場合は schemas/value.proto）に Avro スキーマファイルを生成してください。出発点として references/value.avsc を使用し、ユーザーのドメインに合わせて調整してください。Avro を使用する場合にプロデューサー／コンシューマーのコードを生成するときは、references/producer_avro.py（非同期）、references/producer_avro_sync.py（同期）、references/consumer_avro.py（非同期）からコピーしてください。JSON の参照ファイルは使用しないでください。コンストラクターのシグネチャが異なるためです（下記の Avro コンストラクターに関する注意を参照）。

   シリアライザーを作成する前に、スキーマ登録を独立した明示的なステップとして実行してください。sr_client.register_schema() を呼び出す専用の register_schema() 関数を使用し、エラー（認証エラー、ネットワークエラー、権限拒否）は直ちに伝播させてください。登録を裸の try/except で囲まないでください。その後、シリアライザーを auto.register.schemas=False および use.latest.version=True で設定してください。これにより、シリアライザーによる自動登録を確実に防ぎ、アプリケーションの起動時ではなく CI/CD でスキーマを登録する本番環境の慣行に沿うことができます。

   選択したプロデューサー形式とスキーマ形式に適したシリアライザーを使用してください。

   • JSON Schema（デフォルト）： 非同期の場合は confluent_kafka.schema_registry._async.json_schema の AsyncJSONSerializer／AsyncJSONDeserializer、同期の場合は confluent_kafka.schema_registry.json_schema の JSONSerializer／JSONDeserializer を使用します。
   • Avro（WarpStream SR 使用時のデフォルト）： 非同期の場合は confluent_kafka.schema_registry._async.avro の AsyncAvroSerializer／AsyncAvroDeserializer、同期の場合は confluent_kafka.schema_registry.avro の AvroSerializer／AvroDeserializer を使用します。
   • Protobuf（WarpStream SR 使用時の代替）： 非同期の場合は confluent_kafka.schema_registry._async.protobuf の AsyncProtobufSerializer／AsyncProtobufDeserializer、同期の場合は confluent_kafka.schema_registry.protobuf の ProtobufSerializer／ProtobufDeserializer を使用します。

3. 適切なプロデューサー形式を選択してください。 confluent-kafka-python ライブラリには、2 種類のプロデューサー API があります。

   • AsyncIO Producer（confluent_kafka.aio の AIOProducer）：非ブロッキングで、asyncio のイベントループと統合できます。confluent_kafka.schema_registry._async.json_schema の AsyncJSONSerializer および AsyncSchemaRegistryClient と併用してください。すでにイベントループを実行しているアプリケーション（FastAPI、aiohttp、Sanic、asyncio ワーカー）に最適です。
   • 同期 Producer（confluent_kafka の Producer）：配信コールバックを使用したブロッキング呼び出しです。confluent_kafka.schema_registry.json_schema の JSONSerializer および SchemaRegistryClient と併用してください。スクリプト、バッチジョブ、ユーザーがスレッド／プロセスを制御して poll()／flush() を直接呼び出せる最高スループットのパイプラインに最適です。 どちらの形式が用途に合うか、必ずユーザーに確認してください。コンシューマーでは常に AIOConsumer（非同期）を使用します。長時間実行するポーリングループでは非ブロッキング I/O が有効であり、同期／非同期のコンシューマー形式を混在させても、わずかな利点に対して複雑さが増すだけです。

4. 正常なシャットダウン。 非同期プロデューサーは終了前にflush()とclose()の両方を呼び出し、完了を待つ必要があります。同期プロデューサーは終了前にflush()を呼び出す必要があります。そうしないと、バッファー内のメッセージが失われます。コンシューマーはunsubscribe()の後にclose()を呼び出して、不要なリバランスを避け、コンシューマーグループから適切に離脱する必要があります。try/finallyブロックを使用し、KeyboardInterruptやシグナルを処理してください。

5. Confluent Cloud、ローカルのDocker、WarpStreamに対応します。 Confluent Cloudを使用する場合は、SASL_SSLとPLAINメカニズムを設定し、APIキーを.envから読み込んでください。ローカルのDockerを使用する場合は、認証なしでPLAINTEXTを使用してください。WarpStreamを使用する場合は、ユーザーのWarpStreamのデプロイに応じてSASL_SSLまたはPLAINTEXTを使用し、references/warpstream-optimization.mdに記載されているlibrdkafkaのオーバーライド（大きなバッチ、冪等性の無効化、大きなフェッチ、ゾーン対応のclient.id）を適用してください。KAFKA_ENV環境変数（cloud、local、またはwarpstream）で使用する接続方法を切り替えます。すべての設定は.envを介して環境変数から読み込んでください。

6. 実行前に接続を確認します。 プロデューサーまたはコンシューマーを実行する前に、AdminClient.list_topics()を使用して、ブローカーに到達できることとトピックが存在することを確認してください。Schema Registryへの接続は、HTTPヘルスチェックで確認してください。

7. ドメインイベントには必ずメッセージキーを設定します。 エンティティやイベントストリームを表すメッセージ（注文イベント、ユーザーアクション、デバイスのテレメトリ、トランザクションなど）では、producer.produce()にkey=<entity_id>.encode("utf-8")を渡してください。Kafkaはキーに基づいてパーティションを割り当てるため、同じキーを持つメッセージは同じパーティションに入り、順序が保たれます。これは、コンシューマーが順番どおりにイベントを受け取る必要があるOrderCreated → OrderUpdated → OrderCancelledのようなイベントストリームで重要です。各リファレンスファイルのproduce()ヘルパーは、キーとして使用するフィールド名を指定するkey_fieldパラメーターを受け取ります（例：key_field="order_id"、key_field="transaction_id"）。エンティティを識別するフィールドをユーザーに確認し、それをproduce()に渡してください。順序が重要ではないこと（どのパーティションでも構わないステートレスなメトリクスなど）をユーザーが明示した場合に限り、key_field=Noneにしてください。

   WarpStreamの例外： WarpStreamでは、キーをnullにするとスティッキーパーティショニングが有効になり、より大きなバッチが構築され、スループットとコストが大幅に改善されます。ユーザーのユースケースでエンティティごとの順序が不要な場合（独立したテレメトリの測定値、ステートレスなメトリクス、ログなど）は、key_field=Noneを推奨し、スループット上の利点を説明してください。エンティティごとの順序が必要な場合（同じ注文に対するOrderCreated → OrderUpdated → OrderCancelledなど）は、バッチ効率より正確性を優先し、引き続きメッセージキーを設定してください。どちらを使うか判断するため、イベントでキーごとの順序が必要かユーザーに確認してください。

common.py

このモジュールは、設定の読み込みと接続確認を行います。テンプレートとしてreferences/common.pyを使用してください。

producer.py パターン（AsyncIO）

ユーザーがAsyncIO producerを選択した場合は、references/producer.pyをテンプレートとして使用します。

重要なポイント：

• produce()はパラメーターとしてproducerインスタンスを受け取ります。インスタンスを生成することはありません
• producerはmain()で一度だけ作成し、複数回のproduce()呼び出しに渡すことができます
• 非同期シリアライザー（AsyncJSONSerializer）をメッセージに対して呼び出すときは、awaitが必要です
• AIOProducer.produce()は非同期で、asyncio.Futureを返します。メソッドをawaitしてFutureを取得し、次にFutureをawaitして配信済みのMessageを取得する必要があります：future = await producer.produce(...); result = await future
• AIOProducer.flush()とclose()はコルーチンです。finallyブロック内でawaitする必要があります
• シグナルハンドラーは、正常終了のためのシャットダウンイベントを設定します
• スキーマの登録とシリアライザーの作成は別の手順です。register_schema()はスキーマを明示的に登録し、スキーマIDを返します。エラーはただちに伝播します。create_json_serializer()（Avro／Protobufを使用する場合はcreate_avro_serializer()／create_protobuf_serializer()）は、conf={'auto.register.schemas': False, 'use.latest.version': True}を指定してシリアライザーを作成します。
• 形式によってコンストラクター引数の順序が異なるため、schema_registry_clientとschema_strは必ずキーワード引数として渡し、位置引数として渡さないでください。 confluent-kafka-pythonのJSON、Avro、Protobufのシリアライザー／デシリアライザークラスは、位置引数のシグネチャが同じではありません。JSONSerializer／JSONDeserializerでは最初の引数がschema_strですが、AvroSerializer／AvroDeserializerとProtobufSerializer／ProtobufDeserializerでは最初の引数がschema_registry_clientです。形式によって位置引数とキーワード引数を混在させると、TypeError: ... got multiple values for argument 'schema_registry_client'が発生します。パターンを統一するため、すべての形式でキーワード引数を使用してください：
  • await AsyncJSONSerializer(schema_str=schema_str, schema_registry_client=sr_client, conf=conf)
  • await AsyncAvroSerializer(schema_str=schema_str, schema_registry_client=sr_client, conf=conf)
  • await AsyncProtobufSerializer(msg_type=MsgType, schema_registry_client=sr_client, conf=conf)（Protobufではスキーマ文字列の代わりに、生成されたメッセージクラスを指定します） 標準テンプレートについては、references/producer.py（JSON）とreferences/producer_avro.py（Avro）を参照してください。
• AIOProducerのバッチモードではヘッダーはサポートされていません。 AIOProducer.produce()にheaders=を渡さないでください。渡すとNotImplementedErrorが発生します。スキーマの識別はシリアライザーのワイヤーフォーマットプレフィックスによって自動的に処理されます（JSON Schema、Avro、Protobufのシリアライザーが対象です）。詳細は、以下の「ヘッダー内のスキーマIDとワイヤーフォーマット」を参照してください

producer.py パターン（同期）

ユーザーが同期プロデューサーを選択した場合は、references/producer_sync.py をテンプレートとして使用します。

主なポイント：

• produce() はパラメーターとしてプロデューサーのインスタンスを受け取り、自身で作成することはありません
• プロデューサーは main() で一度作成し、複数回の produce() 呼び出しに渡すことができます
• confluent_kafka.schema_registry.json_schema の JSONSerializer（同期）と、confluent_kafka.schema_registry の SchemaRegistryClient を使用します。Avro／Protobuf（例：WarpStream SR と併用）を使用する場合は、confluent_kafka.schema_registry.avro の AvroSerializer または confluent_kafka.schema_registry.protobuf の ProtobufSerializer を使用します
• Producer.produce() は非ブロッキングで、メッセージをキューに追加します。配信コールバックを処理し、内部キューがいっぱいになるのを防ぐため、produce のたびに producer.poll(0) を呼び出します
• バッチ処理後に producer.flush() を呼び出し、処理中のすべてのメッセージが配信されるまでブロックします
• メッセージごとの配信レポートを処理するには、delivery_callback(err, msg) 関数を使用します
• シグナルハンドラーは、正常終了のためのフラグを設定します
• finally ブロック内の flush() によって、バッファー内のメッセージが失われないようにします
• スキーマの登録とシリアライザーの作成は、非同期パターンと同様に別々の手順です。register_schema() はスキーマを明示的に登録します。create_json_serializer()（または Avro／Protobuf を使用する場合は create_avro_serializer()／create_protobuf_serializer()）は、conf={'auto.register.schemas': False, 'use.latest.version': True} を指定してシリアライザーを作成します。どちらもスキーマ ID を返します
• 同期版にも、キーワード引数のみを使用するルールが適用されます。JSONSerializer(schema_str=..., schema_registry_client=..., conf=...)、AvroSerializer(schema_str=..., schema_registry_client=..., conf=...)、または ProtobufSerializer(msg_type=..., schema_registry_client=..., conf=...) のように呼び出します。位置引数の順序は形式によって一貫していないため、位置引数で渡すと TypeError: ... got multiple values for argument 'schema_registry_client' が発生します。テンプレートについては references/producer_sync.py（JSON）および references/producer_avro_sync.py（Avro）を参照してください
• スキーマ ID は、生成するすべてのメッセージで Kafka レコードヘッダー（confluent.value.schemaId）として渡します。これはヘッダーベースのスキーマ識別パターンです。JSON ペイロードを簡潔で読みやすい状態に保てるため、Confluent 以外のコンシューマーでも読み取れます。詳細については、以下の「ヘッダー内のスキーマ ID とワイヤーフォーマット」を参照してください

consumer.py パターン

references/consumer.py をテンプレートとして使用します。

コンシューマーの主なポイント：

• シグナルベースの正常なシャットダウンを行います。unsubscribe() の後に close() を呼び出し、コンシューマーグループから適切に離脱します
• Schema Registry を使用して、AsyncJSONDeserializer（Avro／Protobuf を使用する場合は AsyncAvroDeserializer／AsyncProtobufDeserializer）でデシリアライズします。生データの解析へのフォールバックはなく、Schema Registry が必須です。デシリアライザーはキーワード引数のみで構築してください。例：await AsyncAvroDeserializer(schema_str=schema_str, schema_registry_client=sr_client)。位置引数のシグネチャは形式によって異なり（Avro／Protobuf では schema_registry_client が先、JSON では schema_str が先）、位置引数とキーワード引数を混在させると TypeError: ... got multiple values for argument 'schema_registry_client' が発生します。テンプレートについては references/consumer.py（JSON）および references/consumer_avro.py（Avro）を参照してください
• シャットダウンシグナルを受け取るまで継続的にポーリングするループを使用します

ヘッダー内の Schema ID とワイヤーフォーマット

同期プロデューサー： 参照コードでは、すべてのメッセージで Schema ID を Kafka レコードヘッダー（confluent.value.schemaId）として渡します。これにより JSON ペイロードがクリーンに保たれ（マジックバイトのプレフィックスが付かず）、Confluent 以外のコンシューマーでも読み取ることができ、kcat などのツールでデバッグできます。同期プロデューサーには、この方法を推奨します。

非同期プロデューサー（AIOProducer）： AIOProducer はバッチモードでのカスタムヘッダーをサポートしていません（headers= が渡されると、produce() は NotImplementedError を発生させます）。スキーマの識別には、JSON Schema シリアライザーのワイヤーフォーマットプレフィックス（ペイロードの先頭に付加されるマジックバイトと Schema ID）を使用します。これは既知の制約です。非同期で生成したメッセージにヘッダーを追加しようとしないでください。

使い分け： 下流のコンシューマーがすべて Confluent 対応（Schema Registry デシリアライザーを使用）であれば、どちらの方法も透過的に機能します。下流のコンシューマーに Confluent 非対応（プレーンな JSON コンシューマー）が含まれる場合は、メッセージの値がクリーンな JSON のまま保たれるため、ヘッダーを使って Schema ID を指定する同期プロデューサーが適しています。複数の種類のコンシューマーが混在するエコシステム向けに生成する場合は、このトレードオフを生成する README に記載してください。

schemas/

ユーザーのデータ領域に合ったスキーマファイルを、選択したスキーマレジストリに適した形式で生成します。

JSON Schema（デフォルト）： schemas/value.schema.json に JSON Schema ファイルを生成します。

たとえば、ユーザーが金融取引を生成する場合：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Transaction",
  "description": "A financial transaction event produced to Kafka.",
  "type": "object",
  "properties": {
    "transaction_id": {
      "type": "string",
      "description": "Unique identifier for this transaction."
    },
    "amount": {
      "type": "number",
      "description": "Transaction amount in the specified currency.",
      "default": 0
    },
    "currency": {
      "type": "string",
      "description": "ISO 4217 currency code.",
      "default": ""
    },
    "timestamp": {
      "type": "string",
      "format": "date-time",
      "description": "Time the transaction occurred, in ISO 8601 format."
    },
    "status": {
      "description": "Current state of the transaction.",
      "enum": ["pending", "completed", "failed", "refunded"],
      "default": "pending"
    },
    "metadata": {
      "oneOf": [{"type": "null"}, {"type": "object"}],
      "description": "Optional metadata associated with the transaction.",
      "default": null
    }
  },
  "required": ["transaction_id", "amount", "currency", "timestamp", "status"]
}

Avro（WarpStream SR を使用する場合）： schemas/value.avsc に Avro スキーマファイルを生成します（Protobuf の場合は schemas/value.proto）。同じ金融取引の例を Avro で示すと、次のようになります。

{
  "type": "record",
  "name": "Transaction",
  "namespace": "com.example.kafka",
  "doc": "A financial transaction event produced to Kafka.",
  "fields": [
    {"name": "transaction_id", "type": "string", "doc": "Unique identifier for this transaction."},
    {"name": "amount", "type": "double", "default": 0, "doc": "Transaction amount in the specified currency."},
    {"name": "currency", "type": "string", "default": "", "doc": "ISO 4217 currency code."},
    {"name": "timestamp", "type": "string", "doc": "Time the transaction occurred, in ISO 8601 format."},
    {"name": "status", "type": {"type": "enum", "name": "Status", "symbols": ["pending", "completed", "failed", "refunded"]}, "default": "pending", "doc": "Current state of the transaction."},
    {"name": "metadata", "type": ["null", "string"], "default": null, "doc": "Optional metadata associated with the transaction."}
  ]
}

スキーマ生成ルール

ユーザーのデータ領域に合わせてスキーマを生成または調整する際は、references/schema-generation-rules.md のルールに厳密に従ってください。

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

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

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

ユーザーがローカル Docker を選択した場合は、テンプレートとして references/docker-compose.yml を使用して、必ず docker-compose.yml を生成してください。これにより、単一ノードの Kafka ブローカー（confluentinc/confluent-local を使用）と Confluent Schema Registry が起動します。ユーザーは docker compose up -d を実行するだけで、利用可能な Kafka 環境を構築できます。

重要： confluentinc/confluent-local イメージは KRaft モードを使用し、組み込みのリスナー名として PLAINTEXT（内部、ポート 29092）、PLAINTEXT_HOST（外部、ポート 9092）、CONTROLLER（ポート 29093）を備えています。カスタムのリスナー名を作成してはいけません。イメージの内部設定と競合し、起動ループが発生します。KAFKA_ADVERTISED_LISTENERS と KAFKA_LISTENERS のオーバーライドには、これらのリスナー名を正確に使用してください。Schema Registry が Docker ネットワーク内からブローカーに接続できるように、内部の PLAINTEXT リスナーには localhost ではなく kafka ホスト名を指定する必要があります。

.env.example

ターゲット環境に応じて、適切な .env.example を生成します。

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=python-client
GROUP_ID=python-consumer-group

ローカル Docker：

KAFKA_ENV=local
BOOTSTRAP_SERVER=localhost:9092
TOPIC=demo-topic
SCHEMA_REGISTRY_URL=http://localhost:8081
CLIENT_ID=python-client
GROUP_ID=python-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=python-client,ws_az=us-east-1a
GROUP_ID=python-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

requirements.txt

JSON Schema（デフォルト）：

confluent-kafka[json,schema_registry]>=2.13.2
jsonschema
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

Avro（例：WarpStream SR を使用する場合）：

confluent-kafka[avro,schema_registry]>=2.13.2
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

Protobuf（例：WarpStream SR を使用する場合）：

confluent-kafka[protobuf,schema_registry]>=2.13.2
python-dotenv
requests>=2.25.0
httpx
authlib
cachetools
attrs
typing_extensions
pytest
pytest-asyncio

生成されたコード（producer.py、consumer.py、common.py）のいずれかでインポートされるサードパーティパッケージはすべて、requirements.txt に対応するエントリを含める必要があります。コードで from confluent_kafka import ... を実行する場合は、requirements.txt に confluent-kafka を含める必要があります。from dotenv import load_dotenv を実行する場合は、python-dotenv を記載する必要があります。これには、自動的にインストールされない推移的依存関係も含まれます。たとえば、非同期 Schema Registry クライアントは実行時に httpx と authlib をインポートするため、これらは confluent-kafka の依存関係として宣言されていなくても、明示的に記載する必要があります。ユーザーが pip install -r requirements.txt を実行してコードを起動したときに、ModuleNotFoundError が一切発生しないようにしてください。

必ず pytest を含めてください。プロジェクトで非同期プロデューサーまたはコンシューマーを使用する場合は、pytest-asyncio も含めてください。プロデューサーが Faker を使ってサンプルデータを生成する場合に限り、Faker を含めてください。

README.md

references/readme-template.md に従って README を生成してください。実際に生成された内容に合わせて調整してください。コンシューマーのみが要求された場合はプロデューサーのセクションを省略し、Confluent Cloud プロジェクトの場合は Docker のセクションを省略してください。

tests/test_project.py

必ずユニットテストを生成してください。テンプレートとして references/test_project.py を使用してください。テストは、稼働中の Kafka クラスターや Schema Registry なしで実行できる必要があります。CI および評価環境でテストが成功するよう、すべての外部依存関係をモックしてください。

テストでは、生成されたコードについて次の項目を検証してください。

1. common.py：load_config() が必須のキーをすべて返し、正しいデフォルト値を使用すること。get_kafka_config() は、KAFKA_ENV=cloud の場合に SASL_SSL と PLAIN を含む設定を生成し、KAFKA_ENV=local の場合には SASL を使わず PLAINTEXT を含む設定を生成すること。モックした処理が成功または失敗したときに、verify_kafka_setup() と verify_schema_registry() が正しい真偽値を返すこと。

2. producer.py（生成された場合）：produce() が producer インスタンスと schema_id をパラメーターとして受け取ること（producer を作成しないこと）。プロデューサークラス（非同期の場合は AIOProducer、同期の場合は Producer）がモジュール内で正確に1回インスタンス化されること。メッセージが生成前にシリアライザーを通ること。生成されるすべてのメッセージで、Kafka レコードヘッダーに confluent.value.schemaId としてスキーマ ID が含まれること。同期プロデューサーの場合は、生成後に flush() が呼び出されることを確認してください。

3. consumer.py（生成された場合）：選択された形式に応じた Schema Registry デシリアライザー（JSONDeserializer／AsyncJSONDeserializer、AvroDeserializer／AsyncAvroDeserializer、または ProtobufDeserializer／AsyncProtobufDeserializer）を使用し、json.loads によるフォールバックを使用しないこと。グレースフルシャットダウンのために、close() の前に unsubscribe() を呼び出すこと。

4. スキーマファイル：テストテンプレートには、TestJsonSchema、TestAvroSchema、TestProtobufSchema の3つのスキーマクラスが含まれています。それぞれ、対応するファイル（value.schema.json、value.avsc、value.proto）の有無を確認する pytest.mark.skipif で保護されています。選択された形式のクラスは実行され、その他はスキップされます。JSON Schema の検証項目：type: object、title、1つ以上のプロパティを含む properties、すべてのプロパティの説明、およびタイムスタンプを表すフィールドの format: date-time。Avro の検証項目：type: record、name、1つ以上のフィールドを含む fields、およびレコードとすべてのフィールドの doc。Protobuf の検証項目：proto3 構文と1つ以上の message 定義。

5. プロジェクト構成：requirements.txt が存在し、confluent-kafka、python-dotenv、requests が含まれていること。.env.example が存在すること。

ユーザーが指定したスキーマとデータのドメインに合わせてテストを調整してください。たとえば、device_id や temperature などのフィールドがある場合、スキーマテストでこれらのフィールド名を確認できます。

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

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

ファイルを生成した後、対象環境に応じた手順をユーザーに案内します。実際に生成された内容に合わせて調整してください。コンシューマーのみが要求された場合はプロデューサーの手順を省略し、プロデューサーのみが要求された場合はコンシューマーの手順を省略します。

Confluent Cloud：

1. .env.example を .env にコピーし、Confluent Cloud の認証情報を入力します
2. virtualenv を作成し、依存関係をインストールします：pip install -r requirements.txt
3. プロデューサーが生成された場合：スキーマは、プロデューサーの初回実行時に register_schema() 関数を通じて明示的に登録されます（auto.register.schemas は False に設定されています）。コンシューマーのみが生成された場合：schemas/value.schema.json の内容を Confluent Cloud Console の Schema Registry > Schemas にある、トピックの value subject に貼り付けて、スキーマを手動で登録します。
4. プロデューサーが生成された場合は実行します：python producer.py
5. コンシューマーが生成された場合は実行します：python consumer.py

ブートストラップサーバー、API キー、Schema Registry URL は、Confluent Cloud Console のクラスターおよび環境の設定から確認できることを伝えてください。

ローカル Docker：

1. Kafka と Schema Registry を起動します：docker compose up -d
2. .env.example を .env にコピーします（ローカル Docker 用のデフォルト値があらかじめ設定されているため、編集は不要です）
3. virtualenv を作成し、依存関係をインストールします：pip install -r requirements.txt
4. トピックの自動作成が無効になっている場合は、トピックを作成します：docker compose exec kafka kafka-topics --create --topic demo-topic --bootstrap-server localhost:29092
5. プロデューサーが生成された場合は実行します：python producer.py
6. コンシューマーが生成された場合は実行します：python consumer.py
7. 完了したら、コンテナを停止します：docker compose down（保存データも削除するには -v を追加します）

WarpStream：

1. .env.example を .env にコピーし、WarpStream のブートストラップサーバー、Schema Registry URL、および該当する場合は認証情報を入力します。ゾーン認識ルーティングを有効にするには、CLIENT_ID に ws_az=<availability-zone> を含めます（例：python-client,ws_az=us-east-1a）
2. virtualenv を作成し、依存関係をインストールします：pip install -r requirements.txt
3. トピックが存在しない場合は作成します：kafka-topics.sh --create --topic demo-topic --bootstrap-server <warpstream-agent>:9092
4. プロデューサーが生成された場合は実行します：python producer.py
5. コンシューマーが生成された場合は実行します：python consumer.py

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