kafka-configuration

事前確認：Spring MCP

このスキルはSpring Agent Toolkitの一部で、Spring MCP サーバー（Amplicode IntelliJ プラグインが提供）と連携するように設計されています。ほかの作業を始める前に、ツール一覧にSpring MCPツールがあるか確認してください。これらのツールはamplicode MCPサーバーの下に公開されています（例：get_project_summary、list_module_dependencies、list_application_properties_files）。MCPツールをツール一覧に展開するハーネスでは、同じ名前の先頭にmcp__amplicode__が付きます。

• Amplicodeツールが1つ以上利用可能な場合 — MCPは接続されています。以下のスキルに進んでください。
• 利用可能なツールがない場合 — 停止し、**amplicode-installスキル（Spring Agent Toolkitに同梱）を実行してください。このスキルはAmplicodeプラグインをインストールし、ウェルカム画面の「Настроить Spring Agent」**ボタンをクリックしてMCPクライアントを再起動する手順を案内します。完了後、MCPツールが利用可能になるので、このスキルを再開してください。
• amplicode-installがスキル一覧に登録されていない場合は、ユーザーの言語で次のように伝えてください："This skill needs the Amplicode IntelliJ plugin and its MCP server. Install it from https://amplicode.ru/marketplace into IntelliJ IDEA Ultimate/Community or GigaIDE, open any project, click «Настроить Spring Agent» on the Amplicode welcome screen, then restart your MCP client."

---

Kafkaの設定

Spring BootのKafkaスターターをapplication.properties／application.yml経由で構成し、必要に応じてKafkaConfigurationクラスを生成します。スターターの依存関係がモジュールのクラスパスに含まれていることを確認します。

重要：コードはexamples/内のファイルのみを使用してください。一致する例がない場合は、停止してユーザーに確認してください。 重要：選択肢が固定されている質問では、AskUserQuestionを優先し、次にその類似ツールを使い、それらがなければ通常のテキストで選択肢を提示してください。 重要：Step 1を実行する前に、会話のコンテキストを読んでください。Step 2〜3の質問の半分は、すでに回答されている可能性があります。

2つの経路（Step 3で選択）

• 経路A — serializerSource = properties（デフォルト）。6つのspring.kafka.*キーをすべてプロパティソースに設定します：bootstrap-servers、consumer.group-id、プロデューサーのキー／値シリアライザー、コンシューマーのキー／値デシリアライザー。KafkaAutoConfigurationがこれらを基にProducerFactory、ConsumerFactory、KafkaTemplate（Bean kafkaTemplate）、kafkaListenerContainerFactoryを作成します。
• 経路B — serializerSource = @Configuration。プロパティソースに設定するのはbootstrap-serversとconsumer.group-idのみです。4つのBeanを持つKafkaConfigurationクラスが生成されます：{prefix}ProducerFactory、{prefix}KafkaTemplate、{prefix}ConsumerFactory、kafkaListenerContainerFactory。{prefix}はproducerValueTypeの単純名を小文字にしたものです。リスナーファクトリのBean名は固定です — autoconfigは@ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")によって後退しますが、別の名前を指定するとautoconfigのファクトリが残り、@KafkaListenerは誤ったファクトリへ黙ってルーティングされます。

既定値

オプション	既定値	必ず確認するか
producerKeyType / producerValueType / consumerKeyType / consumerValueType	java.lang.String	はい — producer のキーと値は必ず確認します。consumer は対称設定が既定です
consumerGroup	—	はい（必須）
bootstrapServers	localhost:9092	いいえ
serializerSource	properties	はい
className	KafkaConfiguration	いいえ（パス B の場合のみ）
packageName	mainPackage	いいえ（パス B の場合のみ）
language / bootVersion	get_project_summary の結果	いいえ

スマートな既定値。 ユーザーが「既定値を使用」「すべて既定値」「最小限の設定」と指定した場合は、Always ask = NO の質問をすべて省略します。必須項目（consumerGroup、serializerSource、およびユーザーがすでに既定値以外を指定した型）のみを確認します。

スマートな回答の認識。 ユーザーが値を直接指定した場合（「OrderEvent メッセージ用」「グループは orders」など）は、再度確認せずに受け入れます。1 つのメッセージに複数の回答がある場合は、すべて受け入れます。

質問のまとめ。 関連する質問を 1 回の AskUserQuestion 呼び出しにまとめます（最大 4 問）。推奨する選択肢を最初にし、ラベルに「（推奨）」を付けます。

意思決定 — まずコンテキストを確認し、その後に質問する

入力ごとに、get_project_summary、list_module_dependencies、list_application_properties_files、過去のターン、ユーザーのプロンプトから値を導き出せるか確認します。コンテキストから明確な既定値が得られない場合にのみ質問します。

1. コンテキストから一意に判断できる → 質問せずに決定します。 language、bootVersion、mainPackage、bootstrapServers = localhost:9092、単一ファイルの propsFile。
2. 強い根拠がある → 1 行で確認します。 決定内容と代替案を示します。ユーザーは何も返答せずにそのまま受け入れられます。
3. 明確な既定値がない → 推奨する選択肢を最初にして AskUserQuestion を呼び出します。
4. 重要な入力が空欄 → そのまま質問します。 consumerGroup、serializerSource、カスタム型の FQN。

ステップ 0 — 会話のコンテキスト（頭の中で行い、ツール呼び出しはしない）

ユーザーのプロンプトと過去のターンを読み直し、すでに示された内容を確認します。

入力	プロンプト内の手がかり
producerValueType / producerKeyType / consumerValueType / consumerKeyType	「X を publish」「X を送信」「キーは Y」「X を consume」「@KafkaListener が X を受信」。consumer は producer と対称の設定が既定です。
consumerGroup	「グループは X」「consumer group id …」
serializerSource	「プロパティを使用」「@Configuration クラス内で」「プロパティのみ」「bean クラス」
className / packageName	「名前は FooConfig」「package は …」
language	Java／Kotlin／ファイル拡張子
スマートな既定値	「既定値を使用」「すべて既定値」→ すべての型を java.lang.String に設定し、consumerGroup と serializerSource のみを確認します

該当する項目にチェックを付け、対応する質問を省略します。ステップ 0 についてユーザーに伝えてはいけません。

ステップ 1 — コンテキストを収集する（MCP 呼び出しとファイル読み取りを並行して実行）

ツール	変数
get_project_summary	language、bootVersion、mainPackage、buildFile、modules リスト
list_module_dependencies(moduleName)	presentDeps
list_application_properties_files(moduleName)	propsFiles
list_spring_beans(moduleName)	existingBeans

呼び出し後、propsFiles 内の各パスを Read し、既存の spring.kafka.* キーを解析して existingProps に格納します。

マルチモジュールの選択。 modules に複数のエントリーがある場合：

1. 各モジュールについて、list_module_dependencies と list_application_properties_files を並行して実行します。
2. 各モジュールをスコアリングします。依存関係に org.springframework.boot:spring-boot-starter-kafka が含まれていれば +1、プロパティファイルに spring.kafka.* キーがあれば +1。
3. スコアが 1 以上のモジュールがちょうど 1 つの場合 → moduleName として選択し、ユーザーには確認しません。
4. 2 つ以上、またはすべて 0 の場合 → ユーザーにモジュールを尋ねます。その後、選択されたモジュールに対してモジュール単位のツールを再実行します。

導出する値：

• bootBranch — boot3（Boot 3.0/3.1）、boot3.2（3.2/3.3/3.4/3.5）、boot4（≥4.0）。ステップ 5 とステップ 6 で使用します。
• kafkaAlreadyPresent — presentDeps に org.springframework.boot:spring-boot-starter-kafka が含まれているかどうか。true の場合はステップ 5b をスキップします。
• singlePropsFile — len(propsFiles) == 1。true の場合はステップ 3 のプロパティファイルに関する質問をスキップします。
• existingBootstrapServers — existingProps に spring.kafka.bootstrap-servers がある場合はその値、ない場合は null。ステップ 2 で暗黙的に使用します。
• existingConsumerGroup — existingProps に spring.kafka.consumer.group-id がある場合はその値、ない場合は null。
• existingKafkaConfig — KafkaTemplate、ProducerFactory、ConsumerFactory、*ListenerContainerFactory Bean のいずれかを宣言する @Configuration クラスに属する existingBeans のエントリー。各エントリーには、クラスの FQN、ソースファイルのパス、すでに宣言されている Bean 名が含まれます。ステップ 4 で使用します。

ステップ 2 — 質問をすべて1回にまとめる

すべての質問を単一の AskUserQuestion 呼び出しで尋ねます（最大 5 件）。コンテキストからデフォルト値を事前入力し、すでに回答済みの質問はスキップします。

1. プロデューサーのキー型は？ — 選択肢：java.lang.String（推奨）、java.lang.Integer、java.lang.Long、java.util.UUID、java.lang.Void、Custom (specify FQN)
2. プロデューサーの値型は？ — 同じ選択肢
3. コンシューマーのグループ ID は？ — 自由入力、必須。existingConsumerGroup が null でない場合は、暗黙的に事前入力します。
4. ブートストラップサーバーは？ — デフォルトは localhost:9092。existingBootstrapServers が null でない場合は、暗黙的に事前入力します。
5. シリアライザーの指定場所は？ — application.properties (Recommended)（パス A）／@Configuration class（パス B）

コンシューマーの型は、デフォルトでプロデューサーの型と同じにします（対称）。ユーザーが非対称を指定した場合にのみ、別途質問します。

ユーザーが「デフォルトを使用」／「すべてデフォルト」と答えた場合は、型に関する質問をスキップし、4 つすべてを java.lang.String にします。consumerGroup、bootstrapServers、serializerSource のみ尋ねます。

ステップ 3 — クラスの対象（パス B のみ）

existingKafkaConfig が空でない場合は、既存のクラス名を示し、判断原則 2（1 行で確認）を適用します。

プロジェクトには {class.fqn} に Kafka Bean がすでにあります。新しい Bean をそこに追加しますか？（はい／いいえ）

• はい → classTarget = existing。既存クラスの packageName、className、ソースファイルのパスを再利用します。ステップ 5 では、新しいファイルを作成するために Write を使うのではなく、Bean を追加するために Edit を使います。既存の Bean 名と重複する挿入先の Bean 名には、ステップ 5 の規則に従って数値のサフィックスを付けます。
• いいえ（または existingKafkaConfig が空の場合）→ classTarget = new：
  • className — デフォルトは KafkaConfiguration。{packageName}/{className}.{java|kt} とファイル名が衝突する場合は、数値のサフィックスを追加します。
  • packageName — デフォルトは mainPackage。複数の @Configuration パッケージがあり、明確な候補がない場合にのみ尋ねます。

通常はコンテキストから暗黙的に回答します。

ステップ 4 — プロパティを書き込み、依存関係を追加する

4a. プロパティ

必ず次を記述します。

spring.kafka.bootstrap-servers={bootstrapServers}
spring.kafka.consumer.group-id={consumerGroup}

serializerSource = properties の場合は、4 つの（デ）シリアライザーキー（examples/serializer-mapping.md に記載の FQN）も追加します。

spring.kafka.producer.key-serializer={producerKeySerializer}
spring.kafka.producer.value-serializer={producerValueSerializer}
spring.kafka.consumer.key-deserializer={consumerKeyDeserializer}
spring.kafka.consumer.value-deserializer={consumerValueDeserializer}

consumerKeyType または consumerValueType が POJO の場合は、次も追加します。

spring.kafka.consumer.properties[spring.json.trusted.packages]={packages}

{packages} = コンシューマー側の POJO 型のパッケージをカンマ区切りで並べたものです。

consumerKeyType が POJO の場合は、次も追加します。

spring.kafka.consumer.properties[spring.json.key.default.type]={consumerKeyTypeFqn}

consumerValueType が POJO の場合は、次も追加します。

spring.kafka.consumer.properties[spring.json.value.default.type]={consumerValueTypeFqn}

YAML での同等の設定は examples/serializer-mapping.md にあります。既存のキーは上書きし、関係のないキーは削除しないでください。

4b. 依存関係

kafkaAlreadyPresent = true の場合はスキップします。それ以外の場合は、buildFile に org.springframework.boot:spring-boot-starter-kafka を追加します（Boot 3.x と 4.x で同じ artifact id）。続けて refresh_build_system_model を呼び出します。

ステップ 5 — @Configuration を生成する（Path B のみ）

1. examples/configuration-class/{language}-{bootBranch}.md を選択します。
2. テンプレートに記載された変数を置き換えます。
3. {prefix} = decapitalize({producerValueType} simple name) — String → string、OrderEvent → orderEvent、UUID → uUID（先頭の 1 文字だけを小文字にします）。
4. Bean 名の衝突ルール（ステップ 1 の existingBeans と、classTarget = existing の場合は対象クラスで宣言されている Bean を確認します）：
   • {prefix}ProducerFactory／{prefix}KafkaTemplate／{prefix}ConsumerFactory — その名前と完全に一致する Bean がある場合は、数値のサフィックスを追加します（stringProducerFactory_1 など）。
   • リスナーファクトリー：デフォルト名は kafkaListenerContainerFactory です。すでに存在する場合は {prefix}KafkaListenerContainerFactory を使用し、ステップ 6 で、{producerValueType} 用の @KafkaListener に containerFactory = "{prefix}KafkaListenerContainerFactory" を明示的に設定する必要があることを警告します。
5. インデントを判定します。
   • classTarget = existing → 対象ファイルからインデントの例を取得します。
   • classTarget = new → .editorconfig → 既存のソースファイルから例を取得 → それでも取得できない場合は 4 スペースを使用します。
6. FQN の処理：本文中のすべての FQN を短縮名に置き換え、package の後に FQN ごとに 1 行の import を出力します。java.lang と同じパッケージのクラスはスキップします。Java の場合は、サードパーティのブロック（アルファベット順）＋空行＋java.* のブロック（アルファベット順）にします。Kotlin の場合、kotlin.* はサードパーティのブロックに含めます。
7. 書き込み手順：
   • classTarget = new → {module sourceRoot}/{packageName as path}/{className}.{java|kt} にファイル全体を Write します。
   • classTarget = existing → 対象ファイルを Read し、クラスの閉じ波括弧の前に 4 つの @Bean メソッドを追加するよう Edit します。必要な import を既存の import ブロックに統合します（重複なし）。新しいファイルは作成しないでください。

ステップ 6 — レポート

ユーザーとの会話で使われている言語に合わせます。次の内容を含めます。

• 実行した Path。
• 書き込みまたは編集したファイル（パス）と、追加したキー／Bean。Path B の場合は、新しいクラスを作成したか、既存のクラスに Bean を追加したかを記載します。
• 依存関係：追加済み／すでに存在。
• 既存のプロジェクト状態から事前入力した値（該当する場合）：«既存の spring.kafka.bootstrap-servers={existingBootstrapServers}／consumer.group-id={existingConsumerGroup} を {propsFile} から再利用しました — 新しい値を指定した場合に限り上書きしました。»
• 使用方法：KafkaTemplate<{producerKeyType}, {producerValueType}> を注入します（Path A では Bean kafkaTemplate、Path B では {prefix}KafkaTemplate）。リスナーファクトリー Bean の名前が kafkaListenerContainerFactory と完全に一致する場合、@KafkaListener はデフォルトでその Bean を使用するため、containerFactory を明示的に指定する必要はありません。
• リスナーファクトリー名にサフィックスを付けた場合：{producerValueType} 用の @KafkaListener に containerFactory = "{prefix}KafkaListenerContainerFactory" を明示的に設定する必要があることを警告します。

ハルシネーション防止チェックリスト

• (De)serializer の FQN は、記憶に頼らず、4 つの型パラメーターそれぞれに一致する examples/serializer-mapping.md の行から取得します。
• serializerSource にかかわらず、bootstrap-servers と consumer.group-id を記述します。
• パス A：4 つのシリアライザーキーはプロパティソースに含めます。@Configuration は生成しません。
• パス B：4 つのシリアライザーキーはプロパティソースに含めません。生成クラス内に記述します。
• リスナーファクトリーの Bean 名は、正確に kafkaListenerContainerFactory とします（デフォルト名が前の生成ですでに使用されている場合に限り、{prefix}KafkaListenerContainerFactory とします）。
• リスナーファクトリーの戻り値の型は ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}> とします。raw 型は使用しません。
• KafkaProperties のインポートは bootBranch に合わせます。boot3／boot3.2 では org.springframework.boot.autoconfigure.kafka、boot4 では org.springframework.boot.kafka.autoconfigure を使用します。
• boot3.2 では buildProducerProperties(sslBundles.getIfAvailable())／buildConsumerProperties(sslBundles.getIfAvailable()) を呼び出します。boot3 と boot4 では引数を渡しません。
• コンシューマー型が POJO の場合：spring.json.trusted.packages（パス A）または JsonDeserializer.TRUSTED_PACKAGES／JacksonJsonDeserializer.TRUSTED_PACKAGES（パス B）には、該当するパッケージを設定します。* や FQN は使用しません。
• consumerKeyType が POJO の場合：spring.json.key.default.type（パス A）または JsonDeserializer.KEY_DEFAULT_TYPE／JacksonJsonDeserializer.KEY_DEFAULT_TYPE（パス B）に、キー型の FQN を設定します。
• consumerValueType が POJO の場合：spring.json.value.default.type（パス A）または JsonDeserializer.VALUE_DEFAULT_TYPE／JacksonJsonDeserializer.VALUE_DEFAULT_TYPE（パス B）に、値型の FQN を設定します。
• インデントは検出したプロジェクトのスタイルに合わせます。