# securitymark_openresty

**Repository Path**: composer-liu-ming/securitymark_openresty

## Basic Information

- **Project Name**: securitymark_openresty
- **Description**: 基于openresty实现标识注入
- **Primary Language**: Lua
- **License**: Apache-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 2
- **Created**: 2026-04-30
- **Last Updated**: 2026-06-30

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# OpenResty 反向代理开发模版

通用 OpenResty 反向代理开发框架，支持 JSON 水印嵌入（键顺序编码）、响应头/响应体修改，开箱即用。

---

## 目录结构

```
securitymark_openresty/
├── shared/                        # 可复用的共享资源
│   ├── mime.types
│   └── lua/
│       ├── watermark.lua          #   JSON 水印嵌入/提取（64位键顺序编码）
│       ├── mark.lua               #   x-mark 安全标记生成（8字段管道格式）
│       ├── store.lua              #   x-mark/水印 Kafka 日志记录（仅身份）
│       ├── traffic_log.lua        #   请求/响应全量 Kafka 日志记录（仅上下文）
│       ├── bch.lua                #   BCH(64,48) 编码器（已废弃，resty.random 替代）
│       ├── analyzer.lua           #   响应体分析器（JSON/HTML 自动识别）
│       ├── header.lua             #   Header 修改工具（增/删/改/条件）
│       ├── body.lua               #   响应体修改工具（追加/替换/JSON变换）
│       ├── utils.lua              #   通用工具函数
│       └── resty/                 #   Kafka Lua 客户端（lua-resty-kafka）
│           └── kafka/
├── data/                          # 运行时数据（gitignore）
├── proxies/                       # 各个代理项目
│   ├── baidu-mark/                #   百度代理（水印+x-mark，每请求随机水印）
│   │   ├── lua/my_handler.lua     #       处理逻辑：A/B/C 模式
│   │   └── reference/             #       mark.ini（已废弃，水印改为 resty.random）
│   ├── github-test/               #   GitHub 代理（仅水印，硬编码 HUST_123）
│   │   ├── lua/my_handler.lua     #       处理逻辑：A 模式
│   │   └── reference/mark.ini     #       64位二进制水印文件
│   ├── dataflow-mark/             #   dataflow 代理（仅 x-mark）
│   ├── wrjtest/                   #   wrjtest SPA 代理（both 模式）
│   ├── platform-12006/            #   平台代理 12006（both 模式）
│   ├── platform-2504/             #   平台代理 2504（both 模式）
│   ├── response-analyzer/         #   响应体分析代理（8081）
│   └── extract.html               #   水印提取工具（浏览器端）
├── template/                      # 新代理项目脚手架（三选一 cp）
│   ├── README.md
│   ├── watermark-only/            #   Pattern A: 仅水印
│   │   ├── nginx.conf             #     含 HTTP + HTTPS 双 server（YOUR_* 占位符）
│   │   ├── lua/my_handler.lua
│   │   ├── ssl/                   #     自签 localhost 证书（开箱即用）
│   │   └── logs/
│   ├── xmark-only/                #   Pattern B: X-Mark + Classify（无 Store）
│   │   ├── nginx.conf
│   │   ├── lua/my_handler.lua
│   │   ├── ssl/
│   │   └── logs/
│   └── both/                      #   Pattern C: 水印 + X-Mark + Classify + Store
│       ├── nginx.conf
│       ├── lua/my_handler.lua
│       ├── ssl/
│       └── logs/
└── scripts/
    ├── proxyctl.sh                # 统一管理脚本
    ├── kafka-consumer.sh          # Docker 基础设施一键管理脚本
    ├── docker-compose.yml         # MySQL / Kafka / Zookeeper / Consumer 容器
    ├── init.sql                   # MySQL 初始化建表
    ├── Dockerfile.consumer        # Consumer Docker 镜像
    └── kafka_consumer.py          # Kafka 流量消费者
```

---

## 快速开始

### 启动基础设施

运行 Docker 容器（MySQL、Kafka、Zookeeper、Consumer）：

```bash
cd scripts
./kafka-consumer.sh start
```

### 常用命令

```bash
cd scripts

# 查看所有代理及对应端口、运行状态
./proxyctl.sh status

# 列出所有可用代理
./proxyctl.sh list

# 启动 / 重载 / 停止 / 测试
./proxyctl.sh start   baidu-mark
./proxyctl.sh reload  baidu-mark
./proxyctl.sh stop    baidu-mark
./proxyctl.sh test    baidu-mark
```

### 验证

```bash
# 百度代理 - 搜索建议 API（JSON，带水印）
curl "http://localhost:8080/sugrec?ie=utf-8&prod=pc&wd=test"

# GitHub API 代理 - 用户信息（JSON，带水印）
curl http://localhost:8082/users/google

# 百度代理首页（HTML）
curl -I http://localhost:8080/

# 提取水印：浏览器打开 extract.html 页面
# 粘贴 JSON → 点击「提取水印」
```

---

## 已有代理

| 项目 | HTTP | HTTPS | 目标 | Watermark | X-Mark | Classify | Store |
|------|------|-------|------|:---:|:---:|:---:|:---:|
| `baidu-mark` | 8080 | 8445 | www.baidu.com | ✅ | ✅ | ✅ | ✅ |
| `dataflow-mark` | 8099 | — | 222.20.126.197:7999 | — | ✅ | ✅ | — |
| `github-test` | 8082, 8084 | 8443, 8444 | api.github.com, github.com | ✅ | — | — | — |
| `wrjtest` | 8088 | 8446 | 192.168.8.117:8058 | ✅ | ✅ | ✅ | ✅ |
| `platform-12006` | 8089 | 8447 | 222.20.126.55:12006 | ✅ | ✅ | ✅ | ✅ |
| `platform-2504` | 8090 | 8448 | 222.20.126.55:2504 | ✅ | ✅ | ✅ | ✅ |
| `response-analyzer` | 8081 | — | 222.20.126.197:7999 | — | — | — | — |

---

## 数据存储流程

x-mark 与流量日志通过 Kafka 异步写入 MySQL，解耦 nginx 请求路径与数据库写入延迟。

### 架构

```
┌──────────────┐     ┌──────────┐     ┌────────────────┐     ┌───────────────┐
│  nginx proxy  │────▶│  Kafka   │────▶│  kafka_consumer │────▶│    MySQL      │
│ (content_by   │     │          │     │  (Python)       │     │               │
│  _lua_block)  │     │          │     │  Docker 容器     │     │  Docker 容器   │
└──────────────┘     └──────────┘     └────────────────┘     └───────────────┘
       │                   │                                        │
  ┌────┴────┐         ┌────┴────┐                              ┌────┴────┐
  │ store   │──────┬──│xmark_logs│──────────────────────────────│xmark_log│
  │ .save() │      │  └─────────┘                              └─────────┘
  └─────────┘      │
  ┌────────┐       │  ┌────────────┐                            ┌─────────┐
  │traffic │───────┼──│traffic_logs│────────────────────────────│api_logs │
  │_log    │       │  └────────────┘                            └─────────┘
  │.log_   │       │
  │request │       │  ┌──────────────┐      ┌──────────────────┐
  │_resp() │       └──│  配置（异步） │      │xmark_log (审计表) │
  └────────┘          │              │      ├──────────────────┤
                      │ store.lua    │      │id, xmark, uri,   │
                      │ traffic_log  │      │method, status,   │
                      │ .lua         │      │client_ip, ct,    │
                      │              │      │body_size,        │
                      │ brokers:     │      │created_at        │
                      │ 127.0.0.1:   │      ├──────────────────┤
                      │ 9092         │      │api_logs(全量日志) │
                      └──────────────┘      │request,response, │
                                            │performance...    │
                                            └──────────────────┘
```

### 组件说明

| 组件 | 技术栈 | 职责 |
|------|--------|------|
| `store.lua` | OpenResty + lua-resty-kafka | x-mark 记录 → Kafka `xmark_logs` topic |
| `traffic_log.lua` | OpenResty + lua-resty-kafka | 请求/响应全量日志 → Kafka `traffic_logs` topic |
| `kafka_consumer.py` | Python + kafka-python + mysql-connector | 消费两个 topic，按 topic 路由写入对应 MySQL 表 |
| MySQL | mysql:8.0 (Docker) | 持久化存储，`xmark_log` 审计表 + `api_logs` 全量日志表 |

### 启动流程

```bash
# 1. 启动基础设施（Kafka + MySQL + Zookeeper + consumer）
cd scripts && docker compose up -d

# 2. 启动代理
./proxyctl.sh start dataflow-mark

# 3. 测试（JSON 接口会产生 x-mark）
curl http://localhost:8099/api/regions
```

### 查询数据

```bash
# x-mark 审计记录
docker exec -i traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT id, xmark, uri, ct, body_size, created_at FROM xmark_log ORDER BY id DESC LIMIT 10;"

# 按 content-type 统计 x-mark 分布
docker exec -i traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT ct, COUNT(*) FROM xmark_log GROUP BY ct;"

# 全量流量日志
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT * FROM api_logs ORDER BY id DESC LIMIT 5\G"
```

### 已知问题

- Kafka advertised listener 必须使用 IP 地址（`127.0.0.1`）而非 hostname（`localhost`），否则 lua-resty-kafka 的异步 DNS 解析会失败
- 如果 Kafka topic 未自动创建，consumer 首次启动时会报 `Topic xmark_logs is not available`，稍后会自动恢复
- 上游必须关闭压缩（`proxy_set_header Accept-Encoding ""`），否则 `ngx.location.capture()` 拿到的 body 是压缩后的
- MySQL 首次初始化时中文可能因 `character_set_client=latin1` 导致双重编码。`init.sql` 已添加 `SET NAMES utf8mb4;` 修复。手动导入时务必加 `--default-character-set=utf8mb4`

---

## 基础设施管理

> **国内 Docker 镜像加速**：部分机器无法直接拉取 Docker 镜像，请先配置 registry mirror。
> 编辑 `/etc/docker/daemon.json`，添加：
> ```json
> {
>   "registry-mirrors": [
>     "https://docker-0.unsee.tech",
>     "https://docker.tbedu.top",
>     "https://docker.xuanyuan.me"
>   ]
> }
> ```
> 然后 `sudo systemctl restart docker`。

Kafka、MySQL、Zookeeper、Consumer 全部运行在 Docker 容器中。

### 一键管理脚本

`scripts/kafka-consumer.sh` 提供一站式管理：

```bash
cd scripts

# 首次启动（安装 lua-resty-kafka + 启动所有容器）
./kafka-consumer.sh start

# 查看运行状态
./kafka-consumer.sh status

# 重启全部
./kafka-consumer.sh restart

# 停止全部
./kafka-consumer.sh stop

# 查看日志（指定某个服务）
./kafka-consumer.sh logs consumer
./kafka-consumer.sh logs kafka

# 修改 kafka_consumer.py 后重新构建 consumer 镜像
./kafka-consumer.sh build

# 单独安装 lua-resty-kafka（新机器上需要）
./kafka-consumer.sh install-lua
```

首次运行 `start` 会自动：
1. 检测 Docker 环境
2. 下载并安装 `lua-resty-kafka` 到 `shared/lua/resty/`
3. 创建 `docker-compose.yml`、`init.sql`、`Dockerfile.consumer`、`kafka_consumer.py`（仅当文件不存在时）
4. 启动所有容器

### Docker Compose 命令

`kafka-consumer.sh` 是对 `docker compose` 的封装，你也可以直接操作：

```bash
cd scripts

# 启动全部
docker compose up -d

# 启动单个服务
docker compose up -d kafka

# 停止全部
docker compose down

# 查看状态
docker compose ps

# 查看日志
docker compose logs -f --tail=50 consumer

# 重新构建并重启 consumer（代码修改后）
docker compose build consumer
docker compose up -d consumer

# 重启 Kafka（修改配置后）
docker compose restart kafka
```

### Kafka UI

Kafka-UI 提供 Web 界面管理 topic 和查看消息：

- URL: http://localhost:8081
- 功能：查看 topic、消费消息、监控 consumer group

### 常见操作场景

```bash
# 场景 1：修改了 kafka_consumer.py → 重建 consumer
./kafka-consumer.sh build

# 场景 2：修改了 docker-compose.yml（如 Kafka 配置）→ 重启对应服务
docker compose restart kafka

# 场景 3：从零开始部署
git clone <repo> && cd scripts
./kafka-consumer.sh start
./proxyctl.sh start dataflow-mark
```

---


### 配置示例

推荐使用 `content_by_lua` + `ngx.location.capture()` 模式（支持在响应发出前设置 header）：

```nginx
init_by_lua_block {
    local p = ngx.config.prefix()
    package.path = p .. "../../shared/lua/?.lua;" .. package.path
    require("watermark").set_watermark("你的64位二进制水印")
    local store = require("store")
    store.config({ topic = "xmark_logs" })
    store.init()
}

server {
    location /upstream/ {
        internal;
        proxy_pass https://目标域名/;
        proxy_ssl_server_name on;
        proxy_set_header Host             目标域名;
        proxy_set_header Accept-Encoding  "";
    }
    location / {
        content_by_lua_block {
            require("my_handler").process()
        }
    }
}
```

> x-mark 记录通过 Kafka 异步写入 MySQL（consumer 在 `scripts/kafka_consumer.py`）。

具体逻辑在 `lua/my_handler.lua` 中实现（参考 `proxies/baidu-mark/lua/my_handler.lua`）。

### 注意事项

- 上游必须关闭压缩（`proxy_set_header Accept-Encoding "";`），否则 `ngx.location.capture()` 拿到的 body 是压缩后的
- 水印仅对 `Content-Type: application/json` 或 `application/vnd*` 的响应生效
- 如果上游 API 返回 `text/plain` 但内容是 JSON，需在 `my_handler.lua` 中改写 Content-Type（参考 baidu-mark）
- `content_by_lua` 模式会完全缓冲响应体，不适合超大文件或流式接口（SSE）

---

## 共享模块说明

| 模块 | 主要函数 | 推荐时机 | 用途 |
|------|----------|----------|------|
| `watermark.lua` | `.set_watermark()`, `.set_external_watermark()`, `.transform()`, `.get_current()`, `.generate()` | init + content_by_lua | JSON 水印嵌入/提取（64位键顺序编码） |
| `mark.lua` | `.compute()`, `.set_header()`, `.config` | content_by_lua | x-mark 安全标记生成（8字段管道格式） |
| `store.lua` | `.init()`, `.config()`, `.save(xmark, {proxy_name, watermark})` | init + content | xmark/水印身份 → Kafka xmark_logs 表 |
| `classify.lua` | `.config()`, `.init()`, `.classify()` | init_by_lua + init_worker timer | 从 MySQL 加载分类规则，计算数据密级（1-5），供 x-mark G 字段使用 |
| `traffic_log.lua` | `.config()`, `.log_request_response()`, `.buffer_response_body()`, `.get_status()` | init + body_filter + log | 请求/响应全量日志 → Kafka traffic_logs → api_logs 表 |
| `bch.lua` | `.encode()`, `.extract_data()` | ini | BCH(64,48) 编码器（已废弃，baidu-mark 改用 resty.random） |
| `header.lua` | `.set()`, `.remove_by_pattern()`, `.rename()`, `.set_on_status()`, `.clear_content_length()`, `.copy_upstream_headers()`, `.copy_client_headers()` | header_filter / content / init | 修改响应头 + 请求头透传 |
| `analyzer.lua` | `.inspect()` | body_filter | 分析响应体，追加分析元数据 |
| `body.lua` | `.init()`, `.body.apply()`, `.prepend()`, `.append()`, `.replace_all()`, `.transform_json()` | body_filter + header_filter | 修改响应体 |
| `utils.lua` | `.get_header()`, `.is_content_type()` | 任意阶段 | 通用工具 |
| `resty.random` | `.bytes(len)` | 全局 | OpenSSL RAND_bytes，生成密码学随机字节（系统自带） |

### store.lua 与 traffic_log.lua 的分工

```
store.lua                     traffic_log.lua
  ↓ (content 阶段)              ↓ (log 阶段)
xmark_logs topic              traffic_logs topic
  ↓                             ↓
xmark_log 表                  api_logs 表
┌──────────────────┐          ┌──────────────────────┐
│ proxy_name       │          │ request_uri          │
│ watermark        │          │ request_method       │
│ xmark            │          │ response_status      │
│  仅身份，3列      │          │ request_time_ms      │
│                  │          │ client_ip            │
│                  │          │ request/response body │
│                  │          │  全量上下文，18+列    │
└──────────────────┘          └──────────────────────┘
```

### watermark.lua API

| 函数 | 说明 |
|------|------|
| `.set_watermark(bin64)` | 设置固定水印（github-test 模式，从 mark.ini 读取） |
| `.set_external_watermark(bin64)` | 设置外部生成的水印（baidu-mark 每请求随机模式） |
| `.get_current()` | 返回当前水印 64 位二进制字符串 |
| `.transform(body)` | 对 JSON body 嵌入水印，返回重排键序后的字符串；非 JSON 原样返回 |
| `.generate()` | 生成随机 64 位水印（已被 resty.random 替代） |
| `.embed(data, bin64)` | 对已解码的 Lua table 嵌入水印，返回 ordered keys |
| `.extract(data, keys)` | 从 Lua table 提取水印 |
| `.run()` | body_filter 入口（body_filter 模式下使用） |

### mark.lua

- `mark.config.data_provider` — 设置数据提供者（如 `"www.baidu.com"`）
- `mark.config.content_type_pattern` — 默认 `"json"`，仅匹配到的 Content-Type 才生成 x-mark
- `mark.config.default_classification_level` — 默认 `3`（密级）
- `mark.compute(body, res_headers)` — 返回 8 字段管道格式：`A|B|C|D|E|F|G|H`
  - A = 4位十六进制随机数，B = 数据规模(KB)，C = Unix 时间戳，D = SHA-256 校验和，E = 数据来源，F = 客户端地址，G = 密级，H = SHA-256 签名
- `mark.set_header(xmark_val, resp_headers)` — 注入 x-mark 到响应头

---

## 新增代理项目

> **设计哲学：** 项目只有两个独立功能 — **Watermark（水印）** 和 **X-Mark（安全标记）**。
> 依赖关系：X-Mark → 必须有 Classify；Store → 必须同时有水印和 X-Mark。
> 详见 [CLAUDE.md](CLAUDE.md#design-philosophy--two-independent-features)。

从三个模板中选一个，`cp -r` 然后改占位符即可：

```bash
# Pattern A: 仅水印
cp -r template/watermark-only proxies/my-proxy

# Pattern B: 仅 x-mark（自动带 classify）
cp -r template/xmark-only proxies/my-proxy

# Pattern C: 水印 + x-mark + classify + store
cp -r template/both proxies/my-proxy
```

然后修改 `nginx.conf` 和 `lua/my_handler.lua` 中的占位符：

| 占位符 | 示例 | 说明 |
|--------|------|------|
| `YOUR_PORT` | `8088` | HTTP 监听端口 |
| `YOUR_HTTPS_PORT` | `8443` | HTTPS 监听端口 |
| `YOUR_UPSTREAM` | `https://api.example.com/` | 上游地址 |
| `YOUR_HOST` | `api.example.com` | 上游 Host 头 |
| `YOUR_DATA_PROVIDER` | `api.example.com` | x-mark E 字段 |
| `YOUR_PROXY_NAME` | `my-proxy` | store 身份标识 |

模板已内置 HTTP + HTTPS 双 server、自签 SSL 证书（CN=localhost）、`logs/` 目录。如不需要 HTTPS，注释或删除 HTTPS server block 即可。

```bash
# 启动
./scripts/proxyctl.sh start my-proxy
```

### 功能速查表

| 想要什么 | 模板 | 自动包含 |
|----------|------|----------|
| JSON 键序水印 | `watermark-only` | watermark |
| x-mark 安全标记（带内容分级） | `xmark-only` | mark + classify |
| 水印 + x-mark + 身份记录 | `both` | watermark + mark + classify + store |

---

## HTTPS 支持

模板已内置自签 SSL 证书（`ssl/cert.pem` + `ssl/key.pem`，CN=localhost）和 HTTPS server block，开箱即用。

如需替换为正式证书：

```bash
# 替换证书文件
cp your-cert.pem proxies/my-proxy/ssl/cert.pem
cp your-key.pem  proxies/my-proxy/ssl/key.pem
./scripts/proxyctl.sh reload my-proxy
```

如果需要生成新的自签证书：

```bash
openssl req -x509 -nodes -days 365 -newkey rsa:2048 \
  -keyout proxies/my-proxy/ssl/key.pem \
  -out proxies/my-proxy/ssl/cert.pem \
  -subj "/CN=localhost"
```

> 如果不需要 HTTPS，直接删除或注释掉 `nginx.conf` 中的 HTTPS server block 即可，无需删除 `ssl/` 目录。

---

## 开发调试

### nginx 代理

```bash
# 查看运行时日志（init_by_lua 错误、watermark 生成、Kafka producer 创建等均在此输出）
tail -f proxies/<name>/logs/error.log

# 在 Lua 中加日志
ngx.log(ngx.ERR, "debug: ", some_var)

# 查看代理状态及端口映射
./scripts/proxyctl.sh status

# 列出所有可用代理
./scripts/proxyctl.sh list

# 测试配置语法（修改 nginx.conf 后务必先测试再 reload）
./scripts/proxyctl.sh test <name>
```

### Kafka / MySQL 链路

```bash
# Consumer 日志 — 确认 Kafka 消息是否被消费、MySQL 是否写入成功
docker logs -f --tail=50 traffic-consumer

# 查看 Kafka topic 消息量（确认 nginx 侧 producer 是否发送成功）
docker exec traffic-kafka kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list localhost:9093 --topic traffic_logs --time -1
docker exec traffic-kafka kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list localhost:9093 --topic xmark_logs --time -1

# 消费组 lag — 确认 consumer 是否跟得上 producer
docker exec traffic-kafka kafka-consumer-groups \
  --bootstrap-server localhost:9093 --group traffic-log-group --describe

# 查看最新一条 Kafka 消息（排查 producer 侧问题）
docker exec traffic-kafka kafka-console-consumer \
  --bootstrap-server localhost:9093 --topic traffic_logs \
  --partition 0 --offset latest --max-messages 1 --timeout-ms 5000

# Kafka UI（Web 界面，管理 topic、查看消息）
# http://localhost:8081
```

### MySQL 查询

```bash
# 水印身份记录（xmark_log，3 列）
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT * FROM xmark_log ORDER BY id DESC LIMIT 10;"

# 全量流量日志（api_logs，含请求/响应体）
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT id, request_uri, response_status, request_time_ms, created_at FROM api_logs ORDER BY id DESC LIMIT 5;"

# 按时间范围查询
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT * FROM api_logs WHERE created_at > '2026-06-01' ORDER BY created_at DESC LIMIT 20;"

# 统计请求量
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT request_uri, COUNT(*) as cnt FROM api_logs GROUP BY request_uri ORDER BY cnt DESC;"

# 统计 xmark 各代理分布
docker exec traffic-mysql mysql -utraffic -ptraffic123 traffic_logs \
  -e "SELECT proxy_name, COUNT(*) FROM xmark_log GROUP BY proxy_name;"
```

### 常见调试场景

```bash
# 场景 1: 水印没生效
# 1) 检查 mark.ini 是否存在、内容是否为合法 64 位二进制
cat proxies/<name>/reference/mark.ini
# 2) 看 error.log 中 watermark 加载日志
grep watermark proxies/<name>/logs/error.log
# 3) 确认响应 Content-Type 包含 json

# 场景 2: x-mark / 水印没写入 MySQL
# 1) 检查 Kafka topic offset 是否增长（nginx producer 是否发送成功）
docker exec traffic-kafka kafka-run-class kafka.tools.GetOffsetShell \
  --broker-list localhost:9093 --topic xmark_logs --time -1
# 2) 看 consumer 日志是否有报错
docker logs --tail=30 traffic-consumer
# 3) 确认 store.save() 是否正确调用（检查 my_handler.lua）

# 场景 3: consumer 报错 Schema mismatch
# 修改 kafka_consumer.py 后记得重建镜像
./scripts/kafka-consumer.sh build
```