kafka-streams-programming

Kafka Streams — 架构设计、构建与调试

嵌入 JVM 的流处理库，无需单独的集群。

⚠️ 重要：仅按需加载参考资料

切勿预先读取所有参考文件。只在需要时读取所需内容。

• 用户问“如何联接两个主题？”→ 仅阅读 references/topology-patterns.md 中的 § Joins Decision Tree
• 用户问“帮我构建一个 Kafka Streams 应用”→ 编写构建文件时再阅读 references/build-templates.md，不要提前阅读
• 用户问“我的应用崩溃了”→ 阅读 references/debugging.md 中针对该症状的具体章节
• 大多数问题总共只需查阅 0–2 个参考文件，而不是全部 10 个

切勿为了“以防万一”而预先阅读多个文件

始终先确认目标环境

在任何模式（Architect、Build、Debug）下回答前，如果用户尚未说明目标环境，请先确认：Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream。版本和认证方式会影响每项建议——KIP-1071 支持情况、SASL 配置、ACL 模型、transactional-id 过期时间、CLI 工具名称等都因环境而异。只有当用户已经指定环境时，才跳过此问题。

如果用户选择 WarpStream： 阅读 references/warpstream-optimization.md，并在标准配置基线之上应用其中的覆盖设置。对 Kafka Streams 的主要影响：

• EOS 会显著降低吞吐量——exactly_once_v2 会在内部启用幂等生产者；由于 WarpStream 的在途请求并发能力有限，这会降低吞吐量。除非用户确实需要 EOS，否则默认采用 at_least_once，并在下游进行去重。
• 必须覆盖生产者和消费者配置（更大的批次、更长的 linger、更大的 fetch）。参见共享参考资料中的 “Kafka Streams Specific” 章节。
• 不支持 fetch.min.bytes——不要设置。
• replication.factor 仅为表面配置（始终为 3）——不要调整。
• 通过带有 ws_az=<az> 后缀的 client.id 实现可识别可用区的路由，这对控制成本至关重要。
• 延迟较高（produce 的 p50 约为 250ms，而 Kafka 上为个位数毫秒）——应向用户说明预期。

模式判定

判断用户意图并进入相应模式：

用户意图	模式	操作
“我需要处理主题 X 中的事件……” / “帮我构建一个 KS 应用……” / “我想要聚合／筛选／联接……”	Build	转到 Build Mode
“我应该如何设计拓扑？” / “应该使用 KTable 还是 GlobalKTable？” / “我需要哪种联接类型？” / “如何处理迟到事件？”	Architect	转到 Architect Mode
“我的 Streams 应用卡住了／很慢／崩溃了……” / “为什么会不断重新平衡？” / “如何解读这个指标？”	Debug	转到 Debug Mode

如果意图不明确，默认进入 Architect——先理解问题，再生成代码。

优先考虑 Flink（仅限 Confluent Cloud）

如果用户在没有提及 KS 的情况下询问 Confluent Cloud 上的通用流处理，简要介绍 Flink 作为替代方案即可。不要长篇说教。

---

Architect 模式

设计合适的拓扑。将用户的数据问题转化为 KS 基元。

步骤 1：了解数据问题

先确认目标环境（参见前言）。然后询问（如果用户已回答则跳过）：数据是什么（主题）？输出是什么？输入之间有什么关系（组合／丰富／分组）？

步骤 2：推荐拓扑模式

将问题与模式匹配（仅在需要特定模式时阅读 references/topology-patterns.md）。说明：该模式为何适用、用通俗语言描述数据流、涉及的 KS 基元，以及权衡／替代方案。

关键决策树

需要时，只阅读相关章节：

• 合并主题：references/topology-patterns.md § Joins Decision Tree
• 按时间聚合：references/topology-patterns.md § Windowing Decision Tree
• 丰富数据／查找：references/topology-patterns.md § Enrichment Patterns
• 恰好一次：references/topology-patterns.md § Exactly-Once（推荐前先逐步讲解——如果下游可以去重，至少一次更简单）

用户确认后，进入构建模式。

---

构建模式

生成完整、可运行的 Kafka Streams 项目。

第 1 步：收集需求

提问（已回答的可跳过）：

1. 应用的用途是什么？ 对含糊的需求提出质疑。
2. 主题和数据流：输入／输出主题？是否有 Schema？如果有，获取 Schema（不要生成新的）。主题如何关联（联接／查找／相互独立）？
3. Schema 格式（使用现有 Schema 时跳过）：Avro（默认）| Protobuf | JSON Schema
4. 构建工具：Gradle（默认）| Maven
5. 目标环境（必填）：Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream（生成配置时读取 references/config-baseline.md；如果是 WarpStream，还要读取 references/warpstream-optimization.md 以了解客户端覆盖项）
6. 凭据：CC 需要 2 个 API 密钥（Kafka + SR）。CP/AK 需要 bootstrap 和 SR URL 及身份验证类型（必要时读取 references/cli-commands.md）
7. 部署规模：分区数？实例数？状态大小？（必要时读取 references/architecture.md 或 references/production-hardening.md § Deployment Sizing）
8. 测试数据：已有数据，还是希望生成样例数据？

第 2 步：规划资源

提出计划：要创建的主题（源／输出／DLQ）、要注册的 Schema。KS 会自动创建变更日志／重新分区主题。如果用户表示输入主题已存在，则不要将其纳入 create-topics.sh——脚本只应创建新主题（通常是输出主题和 DLQ）。

第 3 步：生成项目

生成：项目结构、Schema、App.java、TopologyBuilder.java、配置、simplelogger.properties、docker-compose（本地运行时）、脚本、TopologyTest.java、.env.example、监控注释。

仅在需要时阅读参考资料：

• 拓扑代码？→ 查阅 references/topology-patterns.md 中对应的模式
• 构建文件结构？→ 查阅 references/build-templates.md
• Schema 语法？→ 查阅 references/schema-patterns.md
• 配置属性？→ 查阅 references/config-baseline.md 中对应环境的配置块
• 脚本？→ 查阅 scripts/create-topics.sh、scripts/teardown.sh
• 本地开发？→ 查阅 references/docker-compose.md

Gradle：创建构建文件后，运行 gradle wrapper --gradle-version 8.12。

如果用户需要样例数据：生成 SampleDataProducer.java 和 produce 任务。

第 4 步：生产环境加固（目标为生产环境时）

**触发条件：**用户提到“production”／“prod”／“deploy”，或指定 K8s／ECS／Docker Swarm，或要求多个实例。

添加生产环境组件（必要时读取 references/production-hardening.md 了解详情）：Logback JSON 日志、logback.xml、健康检查端点、带 JVM 调优的 Dockerfile、KIP-1034 DLQ 处理器、K8s YAML（如果是 K8s）、shadow／fat jar 插件。

第 5 步：讲解代码

说明拓扑、配置选择、运行方法和监控内容。提及 group.protocol=streams（KIP-1071）可使重新平衡速度提升 50-80%（需要 AK 4.2+/CP 8.2+）。

第 6 步：交接前运行应用

必须使用真实的 Broker 实际启动应用，并观察它进入 RUNNING 状态，然后才能宣布任务完成。 生成的代码即使能够编译，并通过 TopologyTestDriver 测试，启动时仍可能失败——版本不匹配导致的 NoClassDefFoundError、日志记录器静默回退、缺少运行时依赖和导入路径错误，都可能漏过 compile + test，只有连接真实 Broker / Schema Registry 时才会暴露。构建成功不代表应用能够正常运行。

根据第 1 步选择的目标环境，采取相应操作：

本地（通过生成的 docker-compose.yml 运行 Apache Kafka 或 Confluent Platform）：

1. 运行 docker compose up -d，并等待 Kafka + SR 健康就绪（运行 docker compose ps，或 curl SR /subjects）
2. 运行 ./create-topics.sh
3. 在后台启动应用（运行 ./gradlew run 或 mvn exec:java），以便在应用运行时查看日志
4. 跟踪日志，确认约 30 秒内出现 State transition from REBALANCING to RUNNING。如果未看到，请查看实际堆栈跟踪，通过 references/debugging.md § 启动失败进行诊断，修复后重启并再次验证
5. 如果用户需要示例数据：生成几条记录，并确认目标主题上出现输出
6. 停止应用并运行 docker compose down（如果用户想继续迭代，也可以让它保持运行——请先询问）

Confluent Cloud： 通常无法自行端到端运行，因为集群 + SR API 密钥属于用户。尽可能在不使用这些凭据的情况下完成工作，然后将剩余步骤交给用户：

1. 如果 .env 中有真实的 CC 凭据（用户已配置真实的 .env）：在本地运行指向 CC 的应用（./gradlew run 会自动加载 .env），并按上述步骤 3–5 操作。不要仅仅因为这是 CC 就跳过；如果有凭据，就运行应用。
2. 如果凭据是占位符或未提供：不要伪称运行成功。请改为：
   • 运行 ./gradlew build（编译 + 单元测试），并报告结果
   • 列出用户必须运行的确切命令以进行验证（./create-topics.sh --cloud、./gradlew run、references/verification.md § Confluent Cloud 中的消费命令），并说明成功时的表现（出现 State transition from REBALANCING to RUNNING，输出主题上有记录）
   • 明确告诉用户：“我无法使用你的 API 密钥在你的 CC 集群上运行此应用——请运行上述步骤，并将任何错误粘贴回来。”

WarpStream： 通常无法自行端到端运行，因为 WarpStream 集群和凭据属于用户。采用与 Confluent Cloud 相同的方法：

1. 如果 .env 中有真实的 WarpStream 凭据：在本地运行应用（./gradlew run 会自动加载 .env），并按上述本地步骤 3–5 操作。请注意，WarpStream 的元数据延迟较高，因此出现 State transition from REBALANCING to RUNNING 可能需要更长时间。
2. 如果凭据是占位符或未提供：不要伪称运行成功。请改为：
   • 运行 ./gradlew build（编译 + 单元测试），并报告结果
   • 列出用户必须运行的确切命令以进行验证（./create-topics.sh、./gradlew run），并说明成功时的表现（出现 State transition from REBALANCING to RUNNING，输出主题上有记录）
   • 提醒用户在 .env 中设置带有 ws_az=<az> 的 client.id，以启用感知可用区的路由
   • 明确告诉用户：“我无法使用你的凭据在你的 WarpStream 集群上运行此应用——请运行上述步骤，并将任何错误粘贴回来。”

交接时，请明确说明你完成了上述哪些操作。如果运行了应用并看到 RUNNING，就如实说明。如果只进行了编译，就只说明这一点。不要暗示自己进行了实际上并未执行的运行时验证。

如需了解 CC 消费命令、架构感知生产者和重置流程，请阅读 references/verification.md。

---

调试模式

步骤 1：对问题分类

症状	类别	前往
应用启动时崩溃	启动失败	references/debugging.md § 启动失败
应用运行但没有输出／停止处理	处理停滞	references/debugging.md § 处理停滞
反复再平衡／持续再平衡	再平衡	references/debugging.md § 再平衡问题
延迟高／处理缓慢	性能	references/debugging.md § 性能
反序列化错误／毒丸消息	数据质量	references/debugging.md § 反序列化错误
状态存储问题（损坏、增长、恢复）	状态	references/debugging.md § 状态存储问题
线程故障／StreamsUncaughtExceptionHandler	线程健康状况	references/debugging.md § 线程故障
内存问题（OOM、堆使用率高、RocksDB）	内存	references/debugging.md § 内存问题
WarpStream 上吞吐量低或出现 KAFKA_STORAGE_ERROR	WarpStream 配置	references/warpstream-optimization.md

步骤 2：收集上下文

首先确认目标环境（参见前言）——大多数调试路径都取决于环境。然后询问：错误消息、配置、KS／Java 版本，以及这是新应用还是回归问题？

步骤 3：诊断并修复

阅读 references/debugging.md 中与已识别类别对应的章节。提供修复方案并加以说明。

---

不变项检查清单

不可妥协的默认设置。全部应用。仅在需要实现细节时才阅读参考文件。

1. 使用 Schema 的数据：使用 SR 序列化器／反序列化器（SpecificAvroSerde、KafkaProtobufSerde、KafkaJsonSchemaSerde）。设置 schema.registry.url、default.key.serde、default.value.serde。对于 JSON Schema，设置 json.value.type。对于 Protobuf，设置 specific.protobuf.value.type（references/config-baseline.md）
2. 版本：KS 4.x／CP 8.x，Java 17+
3. KIP-1071：group.protocol=streams（默认值）。如果出现 UnsupportedVersionException，则移除。暂不支持：静态成员资格、正则主题、备用副本、预热副本（references/topology-patterns.md § 分配策略）
4. 四层错误处理：DeserializationExceptionHandler、ProcessingExceptionHandler（KIP-1034）、ProductionExceptionHandler、StreamsUncaughtExceptionHandler。对于未捕获的异常，使用 MaxFailures 模式（references/production-hardening.md § 错误处理）
5. 显式命名：ensure.explicit.internal.resource.naming=true
6. 平滑关闭：在收到 SIGTERM／SIGINT 时，通过 Hook 调用 streams.close(30s)
7. 监控：metrics.recording.level=INFO（references/config-baseline.md）
8. 日志详细程度：生成 simplelogger.properties（references/build-templates.md）
9. 防御式拓扑：在生产环境中为 lambda 加上保护（空值检查、try/catch）。开发环境中可以简化。
10. Schema 同等支持：Avro／Protobuf／JSON Schema 均受支持（references/build-templates.md、references/schema-patterns.md）
11. 测试缓存：TopologyTestDriver 测试需要设置 statestore.cache.max.bytes=0，以避免断言结果不确定。
12. Avro 逻辑类型的 Java 映射：Avro 1.12+ 会为 timestamp-millis／timestamp-micros 生成 java.time.Instant，为 date 生成 LocalDate，为 decimal 生成 BigDecimal 等。切勿将原始 long／int 字面量用于生成的 setter 方法——请使用 Instant.EPOCH、Instant.now()、Instant.ofEpochMilli(...)。比较时间戳时，请使用 Instant.isAfter()／isBefore()，而非 Math.max()／Math.min()。这适用于拓扑代码、聚合初始化器、生产者以及测试辅助工具（references/schema-patterns.md § Java 类型映射）。

---

附带脚本

scripts/：create-topics.sh（预先创建主题，--cloud）、teardown.sh（删除主题／状态，--cloud）、produce-test-data.sh（按需生成）。

参考文件（仅按需阅读）

references/topology-patterns.md — 设计、连接、窗口、聚合 | references/architecture.md — 内部机制、容量规划 | references/debugging.md — 故障排查 | references/config-baseline.md — 配置 | references/build-templates.md — 项目结构 | references/schema-patterns.md — Avro/Protobuf/JSON | references/production-hardening.md — 生产环境配置 | references/cli-commands.md — CLI | references/docker-compose.md — 本地开发 | references/verification.md — 检查清单 | references/warpstream-optimization.md — WarpStream 客户端配置覆盖项