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 工具可用 — MCP 已连接。继续执行以下技能。
• 如果没有可用工具 — 停止并调用 amplicode-install 技能（随 Spring Agent Toolkit 一起提供）。该技能会安装 Amplicode 插件，并引导用户点击欢迎屏幕上的 «Настроить Spring Agent» 按钮，然后重启 MCP 客户端。完成后，MCP 工具即可使用 — 继续执行此技能。
• 如果技能列表中未注册 amplicode-install，请用用户的语言告知用户："此技能需要 Amplicode IntelliJ 插件及其 MCP 服务。请从 https://amplicode.ru/marketplace 将其安装到 IntelliJ IDEA Ultimate/Community 或 GigaIDE，打开任意项目，点击 Amplicode 欢迎屏幕上的 «Настроить Spring Agent»，然后重启 MCP 客户端。"

---

Kafka 配置

通过 application.properties / application.yml 配置 Spring Boot 的 Kafka starter，并可选生成 KafkaConfiguration 类。确保 starter 依赖已添加到模块类路径中。

重要：代码只能来自 examples/ 文件。如果没有匹配的示例 — 停止并询问用户。 重要：对于有固定选项的问题，优先使用 AskUserQuestion > 其类似工具 > 纯文本列表。 重要：运行第 1 步前，请先阅读对话上下文。第 2–3 步中的一半问题可能已经得到回答。

两种路径（由第 3 步选择）

• 路径 A — serializerSource = properties（默认）。六个 spring.kafka.* 键都写入属性源：bootstrap-servers、consumer.group-id、producer key/value serializer、consumer key/value deserializer。KafkaAutoConfiguration 会据此创建 ProducerFactory、ConsumerFactory、KafkaTemplate（bean kafkaTemplate）和 kafkaListenerContainerFactory。
• 路径 B — serializerSource = @Configuration。只有 bootstrap-servers + consumer.group-id 写入属性源。生成一个 KafkaConfiguration 类，其中包含四个 bean：{prefix}ProducerFactory、{prefix}KafkaTemplate、{prefix}ConsumerFactory、kafkaListenerContainerFactory。{prefix} 是 producerValueType 的小写简单名称。监听器工厂 bean 名称是固定的 — 自动配置会通过 @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory") 退让；使用其他名称会导致自动配置工厂仍然保留，并静默地将 @KafkaListener 路由到错误的工厂。

默认值

选项	默认值	是否始终询问？
producerKeyType / producerValueType / consumerKeyType / consumerValueType	java.lang.String	是——始终询问生产者的键和内容类型；消费者默认采用对称配置
consumerGroup	—	是（必填）
bootstrapServers	localhost:9092	否
serializerSource	properties	是
className	KafkaConfiguration	否（仅适用于路径 B）
packageName	mainPackage	否（仅适用于路径 B）
language / bootVersion	来自 get_project_summary	否

智能默认值。 如果用户说“使用默认值”/“全部使用默认值”/“最简配置” → 跳过所有“是否始终询问？”为“否”的问题。只询问必填项（consumerGroup、serializerSource，以及用户已提及非默认值的任何类型）。

智能识别回答。 如果用户直接提供值（“针对 OrderEvent 消息”、“分组 orders”），则直接接受，无需再次询问。一条消息中包含多个回答 → 全部接受。

批量提问。 将相关问题合并到一次 AskUserQuestion 调用中（最多 4 个问题）。建议先列出推荐选项，并在其标签中注明“（推荐）”。

决策——先看上下文，再提问

对于每项输入：尝试从 get_project_summary、list_module_dependencies、list_application_properties_files、之前的对话轮次和用户提示中推断。只有当上下文无法得出明确的默认值时才提问。

1. 上下文明确 → 静默决定。 language、bootVersion、mainPackage、bootstrapServers = localhost:9092、单文件 propsFile。
2. 有强烈信号 → 单行确认。 说明决定及备选项；用户可以不作回应，即表示接受。
3. 没有明确的默认值 → 使用 AskUserQuestion，并将推荐选项列在最前面。
4. 关键输入为空 → 直接询问。 consumerGroup、serializerSource、自定义类型的 FQN。

步骤 0——对话上下文（在脑中完成，不调用工具）

重新阅读用户提示和之前的对话轮次；逐项勾选已经说明的内容：

输入	提示中的信号
producerValueType / producerKeyType / consumerValueType / consumerKeyType	“发布 X”、“发送 X”、“键设为 Y”、“消费 X”、“@KafkaListener 接收 X”。消费者默认与生产者对称。
consumerGroup	“分组 X”、“消费者组 ID …”
serializerSource	“使用 properties”、“在 @Configuration 类中”、“仅使用 properties”、“bean 类”
className / packageName	“命名为 FooConfig”、“放在包 … 中”
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

调用完成后，Read propsFiles 中的每个路径，并将任何现有的 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 → 静默将其选为 moduleName。
4. 两个或更多模块得分 ≥ 1，或所有模块得分均为 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 — existingBeans 中属于 @Configuration 类且声明了任何 KafkaTemplate、ProducerFactory、ConsumerFactory、*ListenerContainerFactory Bean 的条目。每个条目包含类 FQN、源文件路径以及已声明的 Bean 名称。在步骤 4 中使用。

步骤 2 — 一次性询问所有问题

在单次 AskUserQuestion 调用中询问所有问题（最多 5 个）。根据上下文预填默认值，跳过已回答的问题：

1. Producer key type? — 选项：java.lang.String（推荐）、java.lang.Integer、java.lang.Long、java.util.UUID、java.lang.Void、Custom (specify FQN)
2. Producer value type? — 相同选项
3. Consumer group id? — 纯文本，必填。如果 existingConsumerGroup 非 null → 静默预填。
4. Bootstrap servers? — 默认值 localhost:9092。如果 existingBootstrapServers 非 null → 静默预填。
5. Where to describe serializers? — application.properties (Recommended)（路径 A）/ @Configuration class（路径 B）

Consumer 类型默认与 Producer 类型相同（对称）。仅当用户表示需要非对称类型时，才单独询问。

如果用户说“使用默认值”/“全部使用默认值”——跳过类型问题，四种类型均默认为 java.lang.String。仅询问 consumerGroup、bootstrapServers、serializerSource。

步骤 3 — 类目标（仅路径 B）

如果 existingKafkaConfig 非空，则应用决策原则 2（一行确认），并指出现有类：

项目已在 {class.fqn} 中包含 Kafka Bean。要在该类中添加新的 Bean 吗？（是/否）

• 是 → classTarget = existing。重用现有类的 packageName、className 和源文件路径。步骤 5 随后使用 Edit 追加 Bean，而不是使用 Write 创建新文件。若插入的 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，还要追加四个（反）序列化器键（FQN 见 examples/serializer-mapping.md）：

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，则跳过。否则，将 org.springframework.boot:spring-boot-starter-kafka 添加到 buildFile（Boot 3.x 和 4.x 使用相同的 artifact id），然后调用 refresh_build_system_model。

步骤 5 — 生成 @Configuration（仅路径 B）

1. 选择 examples/configuration-class/{language}-{bootBranch}.md。
2. 替换模板中列出的变量。
3. {prefix} = decapitalize({producerValueType} simple name) — String → string、OrderEvent → orderEvent、UUID → uUID（仅将首字符转为小写）。
4. Bean 名称冲突规则（检查步骤 1 中的 existingBeans；当 classTarget = existing 时，还要检查目标类声明的 beans）：
   • 对于 {prefix}ProducerFactory / {prefix}KafkaTemplate / {prefix}ConsumerFactory，如果存在名称完全相同的 bean，则追加数字后缀（stringProducerFactory_1，……）。
   • Listener factory：默认名称为 kafkaListenerContainerFactory。如果该名称已存在，则使用 {prefix}KafkaListenerContainerFactory，并在步骤 6 中警告：{producerValueType} 对应的 @KafkaListener 必须显式设置 containerFactory = "{prefix}KafkaListenerContainerFactory"。
5. 检测缩进：
   • classTarget = existing → 从目标文件采样缩进。
   • classTarget = new → .editorconfig → 从任意现有源文件采样 → 回退为 4 个空格。
6. FQN 处理：将主体中的每个 FQN 替换为其短名称；在 package 之后，为每个唯一 FQN 输出一行 import；跳过 java.lang 和同包类。Java 分组：第三方区块（按字母顺序排序）+ 空行 + java.* 区块（按字母顺序排序）。Kotlin：kotlin.* 与第三方区块合并。
7. 写入步骤：
   • classTarget = new → 在 {module sourceRoot}/{packageName as path}/{className}.{java|kt} 写入完整文件。
   • classTarget = existing → 读取目标文件，然后编辑该文件，在类的右花括号前追加四个 @Bean 方法，并将所需 imports 合并到现有 import 区块中（不重复）。不要创建新文件。

步骤 6 — 报告

使用用户对话所用的语言。包括：

• 采用的路径。
• 写入或编辑的文件（路径）以及添加的键/beans。对于路径 B，说明是创建了新类，还是向现有类追加了 beans。
• 依赖项：已添加 / 已存在。
• 从现有项目状态预填的值（如适用）：«已复用现有的 spring.kafka.bootstrap-servers={existingBootstrapServers} / consumer.group-id={existingConsumerGroup}（来自 {propsFile}）——仅当你提供了新值时才用新值覆盖。»
• 使用方法：注入 KafkaTemplate<{producerKeyType}, {producerValueType}>（路径 A 的 bean 为 kafkaTemplate，路径 B 的 bean 为 {prefix}KafkaTemplate）。如果 listener factory bean 的名称恰好是 kafkaListenerContainerFactory，则 @KafkaListener 默认使用它，无需显式设置 containerFactory。
• 如果 listener factory 的名称带有后缀：警告用户，{producerValueType} 对应的 @KafkaListener 必须显式设置 containerFactory = "{prefix}KafkaListenerContainerFactory"。

防幻觉检查清单

• （反）序列化器的 FQN 来自 examples/serializer-mapping.md 中分别匹配四个类型参数的行，而非凭记忆填写。
• 无论 serializerSource 为何，都会填写 bootstrap-servers 和 consumer.group-id。
• 路径 A：属性源中包含四个序列化器键；不会生成 @Configuration。
• 路径 B：属性源中不包含四个序列化器键；这些键位于生成的类中。
• 监听器工厂 Bean 的名称必须是 kafkaListenerContainerFactory（仅当之前生成的配置已占用默认名称时，才使用 {prefix}KafkaListenerContainerFactory）。
• 监听器工厂的返回类型是 ConcurrentKafkaListenerContainerFactory<{consumerKeyType}, {consumerValueType}>，绝不能使用原始类型。
• 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。
• 缩进与检测到的项目风格一致。