kafka-streams-programming

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Kafka Streams — Architect, Build, Debug

Kafka Streams — 架构、构建与调试

JVM-embedded stream processing library with no separate cluster.

JVM嵌入式流处理库，无需独立集群。

⚠️ IMPORTANT: Lazy-Load References Only

⚠️ 重要提示：仅按需加载参考文件

Do NOT read all reference files upfront. Read ONLY what you need, when you need it.

User asks "how do I join two topics?" → Read
```
references/topology-patterns.md
```
§ Joins Decision Tree only
User asks "build me a Kafka Streams app" → Read
```
references/build-templates.md
```
when writing build files, not before
User asks "my app is crashing" → Read the specific section in
```
references/debugging.md
```
for that symptom
Most questions need 0-2 reference files total, not all 10

Never read multiple files preemptively "just in case"

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

用户询问“如何连接两个主题？”→ 仅读取
```
references/topology-patterns.md
```
中的「连接决策树」章节
用户询问“帮我构建一个Kafka Streams应用”→ 仅在编写构建文件时读取
```
references/build-templates.md
```
，不要提前读取
用户询问“我的应用崩溃了”→ 读取
```
references/debugging.md
```
中对应症状的特定章节
大多数问题仅需0-2个参考文件，无需全部10个

切勿预先读取多个文件以防万一

Always Confirm Target Environment First

始终先确认目标环境

Before answering in any mode (Architect, Build, Debug), confirm the target environment if the user hasn't stated it: Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream. Versions/auth shape every recommendation — KIP-1071 support, SASL config, ACL model, transactional-id expiry, CLI tool names all branch on this. Skip the question only if the user already named the environment.

If the user selects WarpStream: Read

references/warpstream-optimization.md

and apply its overrides on top of the standard config baseline. Key impacts for Kafka Streams:

EOS has a significant throughput cost —
```
exactly_once_v2
```
enables idempotent producers internally, which reduces throughput on WarpStream due to limited in-flight request concurrency. Default to
```
at_least_once
```
with downstream deduplication unless the user has a strong need for EOS.
Producer and consumer configs must be overridden (larger batches, higher linger, larger fetches). See the "Kafka Streams Specific" section in the shared reference.
```
fetch.min.bytes
```
is not supported — do not set it.
```
replication.factor
```
is cosmetic (always 3) — do not tune it.
Zone-aware routing via
```
client.id
```
with
```
ws_az=<az>
```
suffix is critical for cost.
Latency is higher (~250ms p50 produce vs single-digit ms on Kafka) — set expectations with the user.

在任何模式（架构设计、构建、调试）下作答前，若用户未说明目标环境，请先确认：Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream。版本/认证方式会影响所有建议——KIP-1071支持、SASL配置、ACL模型、事务ID过期时间、CLI工具名称均取决于此。若用户已指定环境，则跳过此问题。

若用户选择WarpStream： 读取

references/warpstream-optimization.md

并在标准配置基准之上应用其覆盖规则。对Kafka Streams的关键影响：

EOS（精确一次语义）会显著影响吞吐量——
```
exactly_once_v2
```
会在内部启用幂等生产者，由于WarpStream的并发飞行请求限制，这会降低吞吐量。除非用户有强烈的EOS需求，否则默认使用
```
at_least_once
```
（至少一次语义）并配合下游去重。
必须覆盖生产者和消费者配置（更大的批次、更高的延迟、更大的抓取量）。请参阅共享参考文件中的「Kafka Streams特定配置」章节。
不支持
```
fetch.min.bytes
```
——请勿设置该参数。
```
replication.factor
```
仅为装饰性参数（始终为3）——无需调优。
通过带有
```
ws_az=<az>
```
后缀的
```
client.id
```
实现区域感知路由对成本控制至关重要。
延迟更高（WarpStream的p50生产延迟约为250ms，而Kafka为个位数ms）——需向用户说明此预期。

Mode Detection

模式检测

Determine the user's intent and enter the appropriate mode:

User intent	Mode	What to do
"I need to process events from topic X..." / "Build me a KS app..." / "I want to aggregate/filter/join..."	Build	Go to Build Mode
"How should I design my topology?" / "Should I use a KTable or GlobalKTable?" / "What join type do I need?" / "How do I handle late events?"	Architect	Go to Architect Mode
"My Streams app is stuck/slow/crashing..." / "Why am I getting rebalancing loops?" / "How do I interpret this metric?"	Debug	Go to Debug Mode

If unclear, default to Architect — understand the problem before generating code.

确定用户意图并进入相应模式：

用户意图	模式	操作
“我需要处理来自主题X的事件...” / “帮我构建一个KS应用...” / “我想要进行聚合/过滤/连接...”	构建模式	前往构建模式
“我应该如何设计拓扑？” / “我应该使用KTable还是GlobalKTable？” / “我需要哪种连接类型？” / “如何处理延迟事件？”	架构设计模式	前往架构设计模式
“我的Streams应用卡住/运行缓慢/崩溃...” / “为什么会出现重平衡循环？” / “如何解读这个指标？”	调试模式	前往调试模式

若意图不明确，默认进入架构设计模式——在生成代码前先理解问题。

Flink-first check (Confluent Cloud only)

Flink优先检查（仅适用于Confluent Cloud）

If user asks for generic stream processing on CC without mentioning KS, briefly offer Flink as alternative. Don't lecture.

若用户在Confluent Cloud上询问通用流处理但未提及KS，可简要提供Flink作为替代方案，无需展开说明。

Architect Mode

架构设计模式

Design the right topology. Translate user's data problem into KS primitives.

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

Step 1: Understand the Data Problem

步骤1：理解数据问题

Confirm target environment first (see preamble). Then ask (skip if answered): What data (topics)? What output? Relationship between inputs (combine/enrich/group)?

首先确认目标环境（见前文）。然后询问（若已回答则跳过）：处理哪些数据（主题）？输出是什么？输入之间的关系（合并/增强/分组）？

Step 2: Recommend the Topology Pattern

步骤2：推荐拓扑模式

Match problem to pattern (read

references/topology-patterns.md

only for the specific pattern needed). Present: why it fits, data flow in plain English, KS primitives involved, tradeoffs/alternatives.

将问题与模式匹配（仅读取

references/topology-patterns.md

中所需的特定模式）。说明：为何该模式适用、用通俗易懂的语言描述数据流、涉及的KS原语、权衡/替代方案。

Key Decision Trees

关键决策树

When needed, read only the relevant section:

Combine topics:
```
references/topology-patterns.md
```
§ Joins Decision Tree
Aggregate over time:
```
references/topology-patterns.md
```
§ Windowing Decision Tree
Enrichment/lookup:
```
references/topology-patterns.md
```
§ Enrichment Patterns
Exactly-once:
```
references/topology-patterns.md
```
§ Exactly-Once (walk through before recommending — at-least-once is simpler if downstream can dedupe)

After user confirms, go to Build Mode.

必要时仅读取相关章节：

合并主题：
```
references/topology-patterns.md
```
中的「连接决策树」章节
按时间聚合：
```
references/topology-patterns.md
```
中的「窗口决策树」章节
数据增强/查询：
```
references/topology-patterns.md
```
中的「增强模式」章节
精确一次语义：
```
references/topology-patterns.md
```
中的「精确一次语义」章节（推荐前先梳理——若下游可去重，至少一次语义更简单）

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

Build Mode

构建模式

Generate a complete, runnable Kafka Streams project.

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

Step 1: Gather Requirements

步骤1：收集需求

Ask (skip if already answered):

What does your app do? Push back on vague requests.
Topics & data flow: Input/output topics? Schematized? If yes, retrieve schema (don't generate new ones). How do topics connect (joins/lookups/independent)?
Schema format (skip if using existing): Avro (default) | Protobuf | JSON Schema
Build tool: Gradle (default) | Maven
Target environment (REQUIRED): Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream (read
```
references/config-baseline.md
```
when generating config; if WarpStream, also read
```
references/warpstream-optimization.md
```
for client overrides)
Credentials: CC needs 2 API keys (Kafka + SR). CP/AK needs bootstrap + SR URLs + auth type (read
```
references/cli-commands.md
```
if needed)
Deployment sizing: Partitions? Instances? State size? (read
```
references/architecture.md
```
or
```
references/production-hardening.md
```
§ Deployment Sizing if needed)
Test data: Has data or wants sample data generated?

询问（若已回答则跳过）：

你的应用要实现什么功能？ 对模糊请求进一步追问。
主题与数据流：输入/输出主题？是否有 schema？若有，获取schema（不要生成新的）。主题之间如何关联（连接/查询/独立）？
Schema格式（若使用现有schema则跳过）：Avro（默认）| Protobuf | JSON Schema
构建工具：Gradle（默认）| Maven
目标环境（必填）：Apache Kafka | Confluent Platform | Confluent Cloud | WarpStream（生成配置时读取
```
references/config-baseline.md
```
；若为WarpStream，还需读取
```
references/warpstream-optimization.md
```
获取客户端覆盖规则）
凭证：Confluent Cloud需要2个API密钥（Kafka + Schema Registry）。Confluent Platform/Apache Kafka需要bootstrap地址 + Schema Registry URL + 认证类型（必要时读取
```
references/cli-commands.md
```
）
部署规模：分区数？实例数？状态大小？（必要时读取
```
references/architecture.md
```
或
```
references/production-hardening.md
```
中的「部署规模」章节）
测试数据：已有数据还是需要生成示例数据？

Step 2: Plan Resources

步骤2：规划资源

Present plan: topics to create (source/output/DLQ), schemas to register. Changelog/repartition topics auto-created by KS. If the user says input topics already exist, omit them from
create-topics.sh
— the script should only create new topics (typically output + DLQ).

展示计划：要创建的主题（源/输出/死信队列DLQ）、要注册的schema。变更日志/重新分区主题由KS自动创建。若用户表示输入主题已存在，请勿在
create-topics.sh
中包含它们——脚本应仅创建新主题（通常为输出主题 + DLQ）。

Step 3: Generate the Project

步骤3：生成项目

Generate: project structure, schemas, App.java, TopologyBuilder.java, config, simplelogger.properties, docker-compose (if local), scripts, TopologyTest.java, .env.example, monitoring comments.

Read references only as needed:

Topology code? →
```
references/topology-patterns.md
```
for the specific pattern
Build file structure? →
```
references/build-templates.md
```
Schema syntax? →
```
references/schema-patterns.md
```
Config properties? →
```
references/config-baseline.md
```
for env-specific blocks

Scripts? →

scripts/create-topics.sh

scripts/teardown.sh

Local dev? →
```
references/docker-compose.md
```

Gradle: Run

gradle wrapper --gradle-version 8.12

after creating build files.

If user wants sample data: generate SampleDataProducer.java and

produce

task.

生成：项目结构、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

任务。

Step 4: Production Hardening (if production target)

步骤4：生产环境加固（若目标为生产环境）

Trigger: User says "production"/"prod"/"deploy" or specifies K8s/ECS/Docker Swarm or requests multiple instances.

Add production components (read

references/production-hardening.md

for details if needed): Logback JSON logging,

logback.xml

, health check endpoint,

Dockerfile

with JVM tuning, KIP-1034 DLQ handler, K8s YAML (if K8s), shadow/fat jar plugin.

触发条件： 用户提及“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插件。

Step 5: Walk Through the Code

步骤5：代码讲解

Explain topology, config choices, how to run, what to monitor. Mention

group.protocol=streams

(KIP-1071) provides 50-80% faster rebalancing (requires AK 4.2+/CP 8.2+).

解释拓扑、配置选择、运行方式、监控要点。提及

group.protocol=streams

（KIP-1071）可将重平衡速度提升50-80%（需要Apache Kafka 4.2+/Confluent Platform 8.2+）。

Step 6: Run the App Before Handing Off

步骤6：交付前运行应用

You must actually start the app against a real broker and observe it reach
RUNNING
before declaring the task done. Generated code that compiles and passes

TopologyTestDriver

tests can still fail at startup — version-mismatch

NoClassDefFoundError

s, silent logger fallbacks, missing runtime deps, and import-path errors all slip past

compile

test

and only surface against a real broker / Schema Registry. A green build is not a working app.

Branch on the target environment chosen in Step 1:

Local (Apache Kafka or Confluent Platform via the generated
docker-compose.yml
):

```
docker compose up -d
```
and wait for Kafka + SR to be healthy (
```
docker compose ps
```
, or curl SR
```
/subjects
```
)
```
./create-topics.sh
```
Start the app in the background (
```
./gradlew run
```
or
```
mvn exec:java
```
) so you can read its logs while it runs
Tail the log and confirm
```
State transition from REBALANCING to RUNNING
```
within ~30s. If you don't see it, read the actual stack trace, diagnose via
```
references/debugging.md
```
§ Startup Failures, fix, restart, re-verify
If the user wanted sample data: produce a few records and confirm output appears on the destination topic
Stop the app and
```
docker compose down
```
(or leave running if the user wants to keep iterating — ask)

Confluent Cloud: You usually cannot run end-to-end yourself because the cluster + SR API keys are the user's. Do the most you can without them, then hand off the rest:

If
```
.env
```
has real CC creds (the user has set up a real
```
.env
```
): run the app locally pointed at CC (
```
./gradlew run
```
auto-loads
```
.env
```
) and follow steps 3–5 above. Don't skip just because it's CC — if you have creds, run it.
If creds are placeholders or not provided: do not fabricate a successful run. Instead:
- Run
```
./gradlew build
```
  (compile + unit tests) and report the result
- List the exact commands the user must run to verify (
```
./create-topics.sh --cloud
```
  ,
```
./gradlew run
```
  , the consume command from
```
references/verification.md
```
  § Confluent Cloud) and what success looks like (
```
State transition from REBALANCING to RUNNING
```
  , records on the output topic)
- Tell the user explicitly: "I couldn't run this against your CC cluster because I don't have your API keys — please run the steps above and paste any errors back."

WarpStream: You usually cannot run end-to-end yourself because the WarpStream cluster and credentials are the user's. Follow the same approach as Confluent Cloud:

If
```
.env
```
has real WarpStream creds: run the app locally (
```
./gradlew run
```
auto-loads
```
.env
```
) and follow the Local steps 3–5 above. Note that
```
State transition from REBALANCING to RUNNING
```
may take longer due to WarpStream's higher metadata latency.
If creds are placeholders or not provided: do not fabricate a successful run. Instead:
- Run
```
./gradlew build
```
  (compile + unit tests) and report the result
- List the exact commands the user must run to verify (
```
./create-topics.sh
```
  ,
```
./gradlew run
```
  ) and what success looks like (
```
State transition from REBALANCING to RUNNING
```
  , records on the output topic)
- Remind the user to set
```
client.id
```
  with
```
ws_az=<az>
```
  in their
```
.env
```
  for zone-aware routing
- Tell the user explicitly: "I couldn't run this against your WarpStream cluster because I don't have your credentials — please run the steps above and paste any errors back."

In the handoff, state plainly which of the above you did. If you ran it and saw

RUNNING

, say so. If you only compiled, say only that. Don't imply a runtime verification you didn't perform.

For CC consume commands, schema-aware producers, and reset procedures, read

references/verification.md

在完成任务前，必须实际启动应用并连接真实代理，观察其进入
RUNNING
状态。可编译且通过

TopologyTestDriver

测试的生成代码仍可能在启动时失败——版本不匹配导致的

NoClassDefFoundError

、静默日志回退、缺失运行时依赖、导入路径错误等问题都能通过

compile

test

，但只有连接真实代理/Schema Registry时才会暴露。编译成功不代表应用能正常运行。

根据步骤1中选择的目标环境进行分支处理：

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

运行
```
docker compose up -d
```
，等待Kafka + Schema Registry健康运行（
```
docker compose ps
```
，或curl Schema Registry的
```
/subjects
```
端点）
运行
```
./create-topics.sh
```
在后台启动应用（
```
./gradlew run
```
或
```
mvn exec:java
```
），以便在运行时查看日志
跟踪日志，确认在约30秒内出现
```
State transition from REBALANCING to RUNNING
```
。若未出现，读取实际堆栈跟踪，通过
```
references/debugging.md
```
中的「启动失败」章节诊断、修复、重启、重新验证
若用户需要示例数据：生成几条记录并确认输出出现在目标主题中
停止应用并运行
```
docker compose down
```
（若用户希望继续迭代则保持运行——需询问用户）

Confluent Cloud： 通常无法自行完成端到端运行，因为集群 + Schema Registry API密钥属于用户。在无密钥的情况下完成力所能及的工作，然后交付剩余步骤：

若
```
.env
```
包含真实的Confluent Cloud凭证（用户已设置真实
```
.env
```
）：在本地运行应用连接Confluent Cloud（
```
./gradlew run
```
会自动加载
```
.env
```
），并按照上述步骤3–5操作。不要因为是Confluent Cloud就跳过——若有凭证，务必运行。
若凭证为占位符或未提供：切勿伪造运行成功的结果。而是：
- 运行
```
./gradlew build
```
  （编译 + 单元测试）并报告结果
- 列出用户必须执行的验证命令（
```
./create-topics.sh --cloud
```
  、
```
./gradlew run
```
  、
```
references/verification.md
```
  中「Confluent Cloud」章节的消费命令）以及成功的标志（
```
State transition from REBALANCING to RUNNING
```
  、输出主题中有记录）
- 明确告知用户：“我无法连接你的Confluent Cloud集群，因为没有你的API密钥——请执行上述步骤并将任何错误信息粘贴回来。”

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

若
```
.env
```
包含真实的WarpStream凭证：在本地运行应用（
```
./gradlew run
```
会自动加载
```
.env
```
），并按照本地环境的步骤3–5操作。注意
```
State transition from REBALANCING to RUNNING
```
可能需要更长时间，因为WarpStream的元数据延迟更高。
若凭证为占位符或未提供：切勿伪造运行成功的结果。而是：
- 运行
```
./gradlew build
```
  （编译 + 单元测试）并报告结果
- 列出用户必须执行的验证命令（
```
./create-topics.sh
```
  、
```
./gradlew run
```
  ）以及成功的标志（
```
State transition from REBALANCING to RUNNING
```
  、输出主题中有记录）
- 提醒用户在
```
.env
```
  中设置带有
```
ws_az=<az>
```
  后缀的
```
client.id
```
  以实现区域感知路由
- 明确告知用户：“我无法连接你的WarpStream集群，因为没有你的凭证——请执行上述步骤并将任何错误信息粘贴回来。”

交付时，明确说明你执行了上述哪项操作。若你运行了应用并看到

RUNNING

状态，请如实说明。若仅编译了代码，也仅说明这一点。不要暗示你未执行的运行时验证。

对于Confluent Cloud的消费命令、支持schema的生产者和重置流程，请读取

references/verification.md

。

Debug Mode

调试模式

Step 1: Classify the Problem

步骤1：问题分类

Symptom	Category	Go to
App crashes on startup	Startup failure	`references/debugging.md` § Startup Failures
App runs but no output / stops processing	Processing stall	`references/debugging.md` § Processing Stalls
Rebalancing loops / constant rebalancing	Rebalancing	`references/debugging.md` § Rebalancing Issues
High lag / slow processing	Performance	`references/debugging.md` § Performance
Deserialization errors / poison pills	Data quality	`references/debugging.md` § Deserialization Errors
State store issues (corruption, growth, recovery)	State	`references/debugging.md` § State Store Issues
Thread failures / `StreamsUncaughtExceptionHandler`	Thread health	`references/debugging.md` § Thread Failures
Memory issues (OOM, high heap, RocksDB)	Memory	`references/debugging.md` § Memory Issues
Low throughput or KAFKA_STORAGE_ERROR on WarpStream	WarpStream config	`references/warpstream-optimization.md`

症状	类别	前往
应用启动时崩溃	启动失败	`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`

Step 2: Gather Context

步骤2：收集上下文

Confirm target environment first (see preamble) — most debug paths branch on it. Then ask for: error message, config, KS/Java versions, new app or regression?

首先确认目标环境（见前文）——大多数调试路径取决于环境。然后询问：错误信息、配置、Kafka Streams/Java版本、是新应用还是回归问题？

Step 3: Diagnose and Fix

步骤3：诊断与修复

Read the relevant section in

references/debugging.md

for the identified category. Provide fix with explanation.

读取

references/debugging.md

中对应类别的相关章节。提供修复方案并说明原因。

Invariant Checklist

不变检查清单

Non-negotiable defaults. Apply all. Read reference files only if you need implementation details.

Schematized data: SR serdes (

SpecificAvroSerde

KafkaProtobufSerde

KafkaJsonSchemaSerde

). Set

schema.registry.url

default.key.serde

default.value.serde

. JSON Schema: set

json.value.type

. Protobuf: set

specific.protobuf.value.type

(references/config-baseline.md)

Versions: KS 4.x / CP 8.x, Java 17+
KIP-1071:
```
group.protocol=streams
```
(default). Remove if
```
UnsupportedVersionException
```
. Unsupported: static membership, regex topics, standby replicas, warm-up replicas (references/topology-patterns.md § Assignment Strategy)
Four-tier error handling:
```
DeserializationExceptionHandler
```
,
```
ProcessingExceptionHandler
```
(KIP-1034),
```
ProductionExceptionHandler
```
,
```
StreamsUncaughtExceptionHandler
```
. Use MaxFailures pattern for uncaught (references/production-hardening.md § Error Handling)

Explicit naming:

ensure.explicit.internal.resource.naming=true

Graceful shutdown: Hook with
```
streams.close(30s)
```
on SIGTERM/SIGINT
Monitoring:
```
metrics.recording.level=INFO
```
(references/config-baseline.md)
Log verbosity: Generate
```
simplelogger.properties
```
(references/build-templates.md)
Defensive topology: Guard lambdas in prod (null checks, try/catch). Dev: simpler is fine.
Schema parity: Avro/Protobuf/JSON Schema all supported (references/build-templates.md, references/schema-patterns.md)
Test caching: TopologyTestDriver tests need
```
statestore.cache.max.bytes=0
```
to avoid non-deterministic assertions.
Avro logical type Java mappings: Avro 1.12+ generates
```
java.time.Instant
```
for
```
timestamp-millis
```
/
```
timestamp-micros
```
,
```
LocalDate
```
for
```
date
```
,
```
BigDecimal
```
for
```
decimal
```
, etc. Never use raw
long
/
int
literals with generated setter methods — use
```
Instant.EPOCH
```
,
```
Instant.now()
```
,
```
Instant.ofEpochMilli(...)
```
. Use
```
Instant.isAfter()
```
/
```
isBefore()
```
instead of
```
Math.max()
```
/
```
Math.min()
```
for timestamp comparisons. Applies to topology code, aggregation initializers, producers, AND test helpers (references/schema-patterns.md § Java type mapping).

不可协商的默认值。全部应用。仅在需要实现细节时读取参考文件。

Schema化数据：使用Schema Registry序列化器/反序列化器（

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

）

版本：Kafka Streams 4.x / Confluent Platform 8.x，Java 17+
KIP-1071：
```
group.protocol=streams
```
（默认）。若出现
```
UnsupportedVersionException
```
则移除。不支持的特性：静态成员资格、正则表达式主题、备用副本、预热副本（参考
```
references/topology-patterns.md
```
中的「分配策略」章节）

四层错误处理：

DeserializationExceptionHandler

、

ProcessingExceptionHandler

（KIP-1034）、

ProductionExceptionHandler

、

StreamsUncaughtExceptionHandler

。对未捕获异常使用MaxFailures模式（参考

references/production-hardening.md

中的「错误处理」章节）

显式命名：

ensure.explicit.internal.resource.naming=true

优雅关闭：在SIGTERM/SIGINT信号上添加
```
streams.close(30s)
```
钩子

监控：

metrics.recording.level=INFO

（参考

references/config-baseline.md

）

日志级别：生成

simplelogger.properties

（参考

references/build-templates.md

）

防御性拓扑：生产环境中的Lambda表达式需添加防护（空值检查、try/catch）。开发环境中可简化。
Schema一致性：支持Avro/Protobuf/JSON Schema（参考
```
references/build-templates.md
```
、
```
references/schema-patterns.md
```
）
测试缓存：TopologyTestDriver测试需设置
```
statestore.cache.max.bytes=0
```
以避免非确定性断言。
Avro逻辑类型Java映射：Avro 1.12+会为
```
timestamp-millis
```
/
```
timestamp-micros
```
生成
```
java.time.Instant
```
，为
```
date
```
生成
```
LocalDate
```
，为
```
decimal
```
生成
```
BigDecimal
```
等。切勿在生成的setter方法中使用原始
long
/
int
字面量——使用
```
Instant.EPOCH
```
、
```
Instant.now()
```
、
```
Instant.ofEpochMilli(...)
```
。比较时间戳时使用
```
Instant.isAfter()
```
/
```
isBefore()
```
而非
```
Math.max()
```
/
```
Math.min()
```
。适用于拓扑代码、聚合初始化器、生产者以及测试辅助工具（参考
```
references/schema-patterns.md
```
中的「Java类型映射」章节）。

Bundled Scripts

捆绑脚本

scripts/

create-topics.sh

(pre-create topics,

--cloud

teardown.sh

(delete topics/state,

--cloud

produce-test-data.sh

(generate if requested).

scripts/

目录包含：

create-topics.sh

（预创建主题，支持

--cloud

参数）、

teardown.sh

（删除主题/状态，支持

--cloud

参数）、

produce-test-data.sh

（若用户请求则生成）。

Reference Files (read on-demand only)

参考文件（仅按需读取）

references/topology-patterns.md

— design, joins, windows, aggregations |

references/architecture.md

— internals, sizing |

references/debugging.md

— troubleshooting |

references/config-baseline.md

— config |

references/build-templates.md

— project structure |

references/schema-patterns.md

— Avro/Protobuf/JSON |

references/production-hardening.md

— prod setup |

references/cli-commands.md

— CLI |

references/docker-compose.md

— local dev |

references/verification.md

— checklists |

references/warpstream-optimization.md

— WarpStream client config overrides

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客户端配置覆盖规则