Apache Kafka Driver — 技术参考

PlantPulse Edge 的 Apache Kafka 驱动是基于 kafka-clients 4.1.2 库封装的 Java 实现。它以 consumer 身份订阅外部 streaming platform（Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API）的 topic，将 record 保存在内存缓存中，并在标签的 polling cycle 时返回缓存值。写入采用 producer.send（异步）。

源码：plantpulse.driver.protocol.kafka.*

类	职责
`KafkaDriver`	`ProtocolDriver` 实现类 —— connect/read/write/close、topic 缓存、lazy subscribe、后台 poller 线程

依赖库：lib/kafka-clients-4.1.2.jar

与 MQTT / WebSocket 的区别

PlantPulse 提供 三种基于消息的驱动 —— 它们都共用同一个 JsonExtract 4-mode JSON 解码器。

驱动	wire protocol	payload	适用场景
`MQTT_CLIENT`	MQTT v3.1.1	任意（通常为 JSON/纯文本）	IIoT broker（HiveMQ / Mosquitto / EMQX），QoS 0/1/2
`WEBSOCKET`	ws/wss	任意（通常为 JSON）	streaming server / 专有 push API
`KAFKA`（本页）	Kafka wire protocol	record key/value（通常为 JSON String）	高吞吐 streaming、replay（offset）、consumer group
`SPARKPLUG_B`	MQTT + Protobuf	Sparkplug B Payload	Cirrus Link / Tahu / Ignition 标准（基于 MQTT 的 IIoT 规范）

Kafka 的核心在于 persistent log + offset replay —— 新的 consumer 以 auto-offset=earliest 注册后即可接收历史数据（在 broker 的 retention 范围内）。

工作流程

broker ──ConsumerRecord──> KafkaConsumer.poll() ──> pollerLoop (background thread)
                                                       │
                                                       ├─ rawByTopic.put(topic, value)
                                                       └─ JsonExtract.updateCache(topicCache, topic, value)

PlantPulse 폴러 ──read(addr)──> JsonExtract.parse(addr.address) ──> JsonExtract.extract(...)   (in-memory hit)
                  write(addr, value) ──KafkaProducer.send(ProducerRecord(topic, value))──> broker

read 首次调用时若 cache miss，则执行 lazy subscribe（调用 consumer.subscribe(...)）后返回空字符串 —— 从下一个 cycle 起才会获得实际值。注册在 tag_map 中的 topic 会在 connect 之后通过 preSubscribeFromTagMap() 批量 subscribe，因此第一个 cycle 就可能收到值。

4-mode JSON 解码规范

JsonExtract 工具类统一了 MQTT / WebSocket / Kafka 三个 driver 的消息解码。

模式	address 格式	行为
SCALAR	`<topic>` 或 `<topic>.value`	将整条消息作为 String。若为 JSON object 则回退为 raw。
KEY	`<topic>:<json-key>`	top-level JSON object 中某个 key 的值
PATH	`<topic>:$.<json-path>`	JSON Pointer 动态求值（例：`$.data.tags.T1` → `/data/tags/T1`）
RAW	`<topic>:_raw_`	最后一条 raw 消息（用于调试）

缓存结构 —— 使用单个 Map<String, String>，key 为 topic + "|" + field：

key	含义
`topic\|_raw_`	最后一条 raw 消息
`topic\|value`	scalar（仅当消息非 JSON 时）
`topic\|<json-key>`	按 top-level JSON key 区分的值（仅当为 object 时）

PATH 模式不经过 cache，每次都对 rawByTopic 的最后一条 raw 消息用 JSONPointer.queryFrom(...) 求值。

选项

键	默认值	说明
`group-id`	`plantpulse-edge-<opc_id>`	consumer group id。offset commit 的单位。
`security-protocol`	`PLAINTEXT`	`PLAINTEXT` / `SSL` / `SASL_PLAINTEXT` / `SASL_SSL`
`sasl-mechanism`	（无）	`PLAIN` / `SCRAM-SHA-256` / `SCRAM-SHA-512`
`sasl-jaas-config`	（无）	JAAS 配置（例：`...PlainLoginModule required username="u" password="p";`）
`auto-offset`	`latest`	`earliest` / `latest` —— 新 group 的起始位置

enable.auto.commit=true 始终被设置，因此 broker 会自动 commit offset（默认周期 5 秒）。

连接生命周期

方法	行为
`connect()`	应用 options → 创建 consumer / producer → 启动 poller 线程 → 批量 subscribe `tag_map`
`pollerLoop()`	循环 `consumer.poll(500ms)` → 对每条 record 执行 `JsonExtract.updateCache(...)`
`read()`	解析 → cache miss 则 lazy subscribe → cache hit 则返回值
`write()`	`producer.send(new ProducerRecord<>(topic, value))`（异步）
`close()`	`consumer.wakeup()` → thread join → 关闭 consumer/producer → 清空缓存

限制与后续工作

重连 —— kafka-clients 自身具备重连逻辑，但当 broker 整体宕机导致 long-lived poll() 失败时，driver 侧不会单独标记 setConnected(false) → 建议通过 broker 侧 metric 监控运行状态。
AWS MSK IAM —— 需要额外添加 aws-msk-iam-auth jar（当前未包含）。后续将考虑把 aws-msk-iam-auth-1.x.jar 放入 lib/ 并将 sasl.client.callback.handler.class 作为选项开放。
TLS truststore —— 无法通过驱动选项指定，使用 JVM 系统 truststore。若 broker 使用自签名私有 CA，请将 CA 证书导入 JVM 的 truststore，或在网关启动选项中添加 -Djavax.net.ssl.trustStore=…。
指定 partition / key 的 producer —— write 始终不带 key 发送，因此分区按 round-robin 决定。也就是说 无法保证同一标签的写入按顺序到达。 若需要分区级顺序，请将 topic 设为单分区，或在 consumer 侧处理顺序保证。
transactional producer —— 不支持。当前也未显式启用 idempotent producer。

测试位置

test/java/plantpulse/driver/protocol/kafka/KafkaDriverTest.java —— 18 个测试。

类加载 / 层次结构（BaseProtocolDriver / ProtocolDriver）
初始状态 / 元信息（isConnected/isWriteSupported/isExternal/getDriverSource）
未连接状态下的 read/write/null 安全性
getDebugBaseUrl —— kafka://、默认端口（9092）
选项解析 —— group-id / security-protocol / sasl-* / auto-offset，options=null 安全性
4-mode 解码 —— 直接填充 cache，验证 KEY / PATH / RAW / SCALAR / .value 后缀全部场景
close() 未连接安全性

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java —— 23 个测试（4-mode parser/cache/extract 集成验证）。

依赖真实 broker 的集成测试另行提供 —— 本单元测试无需外部 Kafka 即可运行。

与 MQTT / WebSocket 的区别​

工作流程​

4-mode JSON 解码规范​

选项​

连接生命周期​

限制与后续工作​

测试位置​