developing-kafka-java-client

在你明确提出并获得问题 #1（现有应用还是从零开始）、#2（目标环境）和 #3（生产者、消费者还是两者）的答案之前，**不要**生成任何代码、搭建任何项目框架或修改任何文件。如果用户的提示只回答了部分问题，仍须先确认你的理解，再生成内容。 无论提示看起来多么具体，此要求均适用。

首先宣布：“Using the Confluent Kafka Java Client skill to guide this project.”

Confluent Kafka Java Client 创建

生成一个可用于生产环境的 Java 项目，使用 Apache Kafka Java 客户端（KafkaProducer / KafkaConsumer）和 Confluent Schema Registry 序列化器向 Kafka 生产消息和/或从 Kafka 消费消息。支持三种目标环境：Confluent Cloud（托管）、Local Docker 和 WarpStream（兼容 Kafka，由对象存储提供支持）；两种构建工具：Maven（默认）和 Gradle；以及三种架构格式：Avro（默认）、JSON Schema 和 Protobuf。生成的代码遵循 Confluent 最佳实践。

第 1 步：收集需求

生成任何代码之前，请先逐一确认以下问题。如果用户已在提示中明确回答了某个问题，请跳过该问题——不要为了走流程而重复询问。例如，“在 Confluent Cloud 上构建生产者和消费者，使用同步发送，并采用 Gradle”已经回答了 #2、#3、#4 和 #5；只剩下 #1、#6、#7、#8、#9 和 #10。

强制确认关卡——不得跳过，即使用户已经回答了所有问题也不例外。 在写入任何文件之前，必须发送一条消息，其中：

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 仓库和序列化器依赖。
   • 如果用户已向 Kafka 生产消息，但未使用 Schema Registry（例如，使用 StringSerializer 和手写 JSON），帮助他们迁移：(1) 根据现有消息结构生成 Avro/JSON 架构；(2) 注册该架构；(3) 将原始 String/byte[] 序列化器替换为 Confluent Schema Registry 序列化器。不要丢弃他们现有的代码。
   • 如果是从头开始，则按下面的完整脚手架流程进行。
2. 目标环境？ — Confluent Cloud、本地 Kafka（Docker）还是 WarpStream。即使用户没有提及，也务必询问。 如果用户提到“本地”、“docker”、“self-hosted”，或只是想在没有云账户的情况下试用 Kafka，则选择本地 Docker。如果用户提到“Confluent Cloud”、“CC”，或已有云凭据，则选择 Confluent Cloud。如果用户提到“WarpStream”，则选择 WarpStream。如果用户确认没有偏好，则默认选择 Confluent Cloud，但务必先询问。
   • **如果选择 WarpStream：**阅读 references/warpstream-optimization.md，并应用该参考文件中的 Java 客户端覆盖配置。主要改动：禁用幂等性，大幅增大批次大小和在途请求数，设置较大的获取大小，并在 client.id 中添加 ws_az=<az> 以支持区域感知路由。除非需要按实体排序，否则优先使用空消息键进行粘性分区。
3. 生产者、消费者，还是两者都要？
4. 发送模式？（仅在用户需要生产者时询问。）Java 的 KafkaProducer.send() 始终会返回 Future<RecordMetadata> — 生产者本质上是异步的。可选的是如何处理结果。帮助用户选择：
   • 使用回调异步处理（推荐的默认选项）：producer.send(record, callback)。不会阻塞；Callback 负责处理每条记录的传递报告和错误。适合大多数应用，吞吐量最高。使用 references/AvroProducer.java。
   • 同步处理：producer.send(record).get()。会阻塞，直到代理确认每条记录。吞吐量最低，但每条消息的确认最可靠，错误处理也最简单。适合批处理/ETL 脚本，以及必须在开始下一步之前确认一次写入的场景。使用 references/AvroProducerSync.java。
   • （为了完整起见，可以提及即发即忘模式 — 调用 producer.send(record)，不带回调，也不调用 .get() — 但不推荐：传递失败会被静默忽略。） 如果用户提到批处理、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、消费者组）时，每个分区只会分配给一个消费者，因此并行度受分区数限制，并保留按键排序。使用 Share Consumer API（KafkaShareConsumer、共享组 — “Kafka 队列”）时，多个共享消费者可以从同一个分区消费，从而提供类似队列的消息传递语义：多个消费者可以协作并发处理消息，而且消费者数量可以超过分区数（代价是无法保证按键排序，并且每条记录都必须确认）。默认使用常规 Consumer API。 当用户需要队列/工作队列语义、竞争消费者、协作式并发处理，或希望消费者扩容不受分区数限制时，选择 Share Consumer API。如果用户选择 Share Consumer API，生成前请阅读 references/share-consumer.md — 它要求使用 Kafka 4.x 客户端（参考构建文件已将 kafka.version=4.2.1 / confluent.version=8.2.1 固定为指定版本），并且需要在集群端启用相应功能。 不要询问是否使用 Schema Registry——始终包含它。默认使用 Avro（这是 JVM 上惯用的选择：生成的 SpecificRecord 类可提供编译时类型安全，并采用紧凑的线缆格式）。如果用户更喜欢 JSON Schema 或 Protobuf，可提供这两种选项——只有在用户暗示偏好时才询问；否则使用 Avro。如果目标是 WarpStream，且用户使用 WarpStream 内置的 Schema Registry，请说明它仅支持 Avro 和 Protobuf（GET /schemas/types 返回 ["AVRO","PROTOBUF"]）——Avro 已是默认选项，因此除非用户想用 Protobuf，否则无需更改。

Agent 常见错误

想法	实际情况
“用户提到了 Spring Boot，所以我会搭建一个全新的 Maven 项目”	如果用户已有应用，应在原项目中修改——将依赖添加到其构建文件，并集成序列化器。不要搭建一个与现有项目竞争的新项目。
“我会像 Python skill 那样默认使用 JSON Schema”	此 skill 在 JVM 上默认使用 Avro。生成的 SpecificRecord 类符合惯用做法且类型安全。只有用户更喜欢时才使用 JSON Schema/Protobuf。
“我会让 auto.register.schemas 保持默认值”	序列化器的默认值是 true。始终将其设为 false，并通过 CachedSchemaRegistryClient.register(...) 显式注册。静默自动注册会带来生产环境风险。
“我会在发送方法中创建一个 KafkaProducer”	在 main() 中创建一个 KafkaProducer 实例，并将其作为参数传入。生产者是线程安全的，而且创建开销很大。
“我会像 Python 同步生产者那样把 schema ID 写入记录头”	Java Confluent 序列化器会自动将 schema ID 嵌入线缆格式（magic byte + 4-byte 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 文件中的值——不要读取该文件。将 .properties 添加到 .gitignore。

核心原则

这些原则非常重要，因为它们能避免使用 Kafka Java 客户端时最常见的生产问题：

1. 复用 producer/consumer 实例。 每条消息都创建一个 KafkaProducer 开销很大——每个实例都会建立 TCP 连接、执行 SASL 握手并获取元数据。KafkaProducer 是线程安全的；创建一个实例并共享它。发送辅助方法会将 producer 作为参数接收，不会自行实例化。

2. 始终使用 Schema Registry，默认采用 Avro。 Schema Registry 用于强制执行 producer/consumer 契约；没有它，schema 变更可能会悄无声息地导致 consumer 出错。本技能默认使用 Avro，因为生成的 SpecificRecord 类能为 JVM 提供编译时类型安全，且线格式紧凑。JSON Schema 和 Protobuf 也是受支持的替代方案。

   在发送消息前，将注册 schema 作为单独的明确步骤执行。 使用专门的 registerSchema() 方法调用 schemaRegistryClient.register(subject, parsedSchema)，并让错误（身份验证失败、网络错误、权限拒绝）立即向上传播——绝不吞掉错误。然后将 serializer 配置为 auto.register.schemas=false 和 use.latest.version=true。这样可确保 serializer 不会悄悄自动注册 schema，并符合生产实践：由 CI/CD 注册 schema，而不是在应用启动时注册。

   使用与所选格式匹配的 serializer：

   • Avro（默认）： io.confluent.kafka.serializers.KafkaAvroSerializer / KafkaAvroDeserializer。在 consumer 上设置 specific.avro.reader=true，使其反序列化为生成的 SpecificRecord 类。
   • JSON Schema： io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer / KafkaJsonSchemaDeserializer。在 consumer 上设置 json.value.type，指定 POJO 类。请参阅 references/JsonSchemaProducer.java 和 references/JsonSchemaConsumer.java。
   • Protobuf： io.confluent.kafka.serializers.protobuf.KafkaProtobufSerializer / KafkaProtobufDeserializer。通过 protobuf-maven-plugin 从 .proto 生成类，并在 consumer 上设置 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。 两种模式下，consumer 都使用单线程 poll() 循环。

4. 优雅关闭。 Producer 必须先执行 flush()，再执行 close()（放在 finally 块中），以免缓冲的记录丢失。Consumer 使用 wakeup 模式：注册 Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup))，运行 poll 循环，直到 consumer.wakeup() 引发 WakeupException；针对性地捕获该异常以退出循环，并在 finally 块中调用 close()。close() 会提交 offset 并让 consumer 干净地退出 consumer group，从而避免不必要的 rebalance。

5. 支持 Confluent Cloud、本地 Docker 和 WarpStream。 对于 Confluent Cloud，通过 sasl.jaas.config 配置 SASL_SSL 和 PLAIN 机制，并从 .properties 文件加载 API key。对于本地 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 显式注册 Schema 并返回 Schema ID；错误会向上传播。序列化器配置为 auto.register.schemas=false、use.latest.version=true。
• Schema 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+ 中，设置 group.protocol=consumer，以使用新一代再平衡协议（KIP-848），该协议可消除全局停顿式再平衡。参见 references/consumer.md。
• 禁用 enable.auto.commit，并在处理后显式提交；根据 references/consumer.md 权衡使用 commitSync 还是 commitAsync。
• 基于 Wakeup 的优雅关闭：关闭钩子调用 consumer.wakeup()；循环捕获 WakeupException，并在 finally 中调用 close()。
• 持续循环调用 poll(Duration.ofMillis(1000))，直到关闭。

共享消费者 API（“Kafka 队列”）。 如果用户在问题 #10 中选择了共享消费者 API，请阅读 references/share-consumer.md，并使用 references/AvroShareConsumer.java，而不是 AvroConsumer.java。它使用 KafkaShareConsumer，让同一共享组中的多个实例协同消费同一分区（类似队列的语义）。与上述相同，使用 Schema Registry 反序列化器和基于 Wakeup 的关闭方式，此外还包括：对每条记录调用 acknowledge(record, AcknowledgeType.ACCEPT/RELEASE/REJECT)，以及在集群侧启用功能（本地设置 share.version=1；在 Confluent Cloud 上启用“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 代理（confluentinc/confluent-local，KRaft 模式）和 Confluent Schema Registry。用户运行 docker compose up -d。

重要： confluentinc/confluent-local 内置的监听器名称为 PLAINTEXT（29092）、PLAINTEXT_HOST（9092）、CONTROLLER（29093）。不要自行定义监听器名称，否则会与镜像的内部配置冲突，导致启动循环。仅使用上述准确名称覆盖 KAFKA_ADVERTISED_LISTENERS 和 KAFKA_LISTENERS。内部 PLAINTEXT 监听器必须将 kafka 主机名设为通告地址，以便 Schema Registry 能在 Docker 网络内访问代理。

.properties.example

为目标环境生成相应的 .properties.example。它是标准的 Java properties 文件（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，并使用 Mockito 模拟 Schema Registry 客户端。

测试验证以下属性：

1. KafkaConfig：当 KAFKA_ENV=cloud 时，baseProperties() 会设置 SASL_SSL + PLAIN（并设置 sasl.jaas.config）；当 KAFKA_ENV=local 时，会设置 PLAINTEXT 且不设置 SASL。针对模拟的 AdminClient，verifyKafkaSetup() 会返回正确的布尔值。
2. 生产者（如果生成）：发送辅助方法接受一个生产者参数（绝不自行创建生产者），类中恰好实例化一个 KafkaProducer，通过 MockProducer 发送一批消息后，每条消息都会在历史记录中留下一个条目，且键符合预期。对于同步变体，验证发送结果已解析（.get()）。
3. 消费者（如果生成）：使用 Schema Registry 反序列化器（KafkaAvroDeserializer / KafkaJsonSchemaDeserializer / KafkaProtobufDeserializer），而不是将原始 StringDeserializer 用于值，并采用 wakeup + close() 关闭模式。使用记录初始化的 MockConsumer 会产生预期的反序列化值。
4. 架构文件：Avro value.avsc 可解析为一个 record，且顶层有 doc，每个字段也都有 doc（参照 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 中手动为该主题的值 subject 注册架构。
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”。使用 mvn exec:java -Dexec.mainClass=com.example.kafka.AvroShareConsumer 运行 share consumer，并启动多个具有相同 GROUP_ID 的实例，即可观察协作式消费。

提醒 WarpStream 用户，其生产延迟（约 250ms p50）高于标准 Kafka，这是预期情况。如果吞吐量偏低，请确认已应用 references/warpstream-optimization.md 中的覆盖设置（尤其是 enable.idempotence=false 和较大的批次/提取大小）。