Apache Kafka ドライバ

PlantPulse Edge の Apache Kafka ドライバは kafka-clients 4.1.2 ライブラリをラップした Java 実装です。外部ストリーミングプラットフォーム (Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API) のトピックを consumer として subscribe してレコードを in-memory キャッシュに保持し、タグのポーリングサイクルごとにキャッシュ値を返します。 write は producer.send (async)。

ソース: plantpulse.driver.protocol.kafka.*

クラス	責任
`KafkaDriver`	`ProtocolDriver` 実装 — connect/read/write/close、トピックキャッシュ、lazy subscribe、background poller thread

ライブラリ: lib/kafka-clients-4.1.2.jar

MQTT / WebSocket との違い

PlantPulse は 3 つのメッセージベースドライバ を提供します — すべて同一の JsonExtract 4-mode JSON デコーダを共有します。

ドライバ	ワイヤプロトコル	payload	使用シーン
`MQTT_CLIENT`	MQTT v3.1.1	任意 (通常 JSON/プレーンテキスト)	IIoT broker (HiveMQ / Mosquitto / EMQX)、QoS 0/1/2
`WEBSOCKET`	ws/wss	任意 (通常 JSON)	ストリーミングサーバ / 独自 push API
`KAFKA` (このページ)	Kafka ワイヤプロトコル	record key/value (通常 JSON String)	高スループットストリーミング、replay (offset)、consumer group
`SPARKPLUG_B`	MQTT + Protobuf	Sparkplug B Payload	Cirrus Link / Tahu / Ignition 標準 (MQTT 上の IIoT 仕様)

Kafka は 永続ログ + 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(...) 呼び出し) 後に空文字列を返す — 次のサイクルから実際の値が入ります。tag_map に登録されたトピックは connect 直後に preSubscribeFromTagMap() で一括 subscribe され、最初のサイクルですぐに値が到着する場合もあります。

4-mode JSON デコード仕様

JsonExtract ユーティリティが MQTT / WebSocket / Kafka の 3 ドライバのメッセージデコードを統一します。

モード	address 形式	動作
SCALAR	`<topic>` または `<topic>.value`	メッセージ全体を String として。JSON object なら raw フォールバック。
KEY	`<topic>:<json-key>`	top-level JSON object の 1 つのキー値
PATH	`<topic>:$.<json-path>`	JSON Pointer 動的 evaluate (例: `$.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 キーごとの値 (object の場合のみ)

PATH モードはキャッシュを経由せず、rawByTopic の最後の raw メッセージを毎回 JSONPointer.queryFrom(...) で evaluate。

オプション

キー	デフォルト	説明
`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` — 新規グループの開始点

enable.auto.commit=true が常に設定され、broker が自動で offset を commit します (デフォルト 5 秒周期)。

接続ライフサイクル

メソッド	動作
`connect()`	options 適用 → consumer / producer 生成 → poller thread 開始 → `tag_map` 一括 subscribe
`pollerLoop()`	`consumer.poll(500ms)` 反復 → record ごとに `JsonExtract.updateCache(...)`
`read()`	parse → cache miss なら lazy subscribe → cache hit なら値を返す
`write()`	`producer.send(new ProducerRecord<>(topic, value))` (async)
`close()`	`consumer.wakeup()` → thread join → consumer/producer close → cache clear

制約と今後の課題

再接続 — kafka-clients は独自の再接続ロジックを持ちますが、broker 全体がダウンして long-lived poll() が失敗する場合、ドライバ側で 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 を使用します。自己署名のプライベート CA を使う broker なら、CA 証明書を JVM の truststore に追加するか、ゲートウェイ起動オプションに -Djavax.net.ssl.trustStore=… を追加してください。
partition / key 指定 producer — write は常に key なしで送信するため、パーティションは round-robin で決まります。つまり 同じタグの書き込みが順序どおりに到着する保証はありません。 パーティション単位の順序が必要な場合は、トピックをパーティション 1 つにするか、順序保証を consumer 側で処理してください。
transactional producer — 未対応。現状 idempotent producer も明示的に enable していません。

テスト場所

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 デコード — キャッシュを直接埋めて KEY / PATH / RAW / SCALAR / .value suffix すべて検証
close() 未接続安全

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java — 23 テスト (4-mode parser/cache/extract 統合検証)。

実 broker 依存の統合テストは別途 — 本ユニットテストは外部 Kafka なしで動作します。

MQTT / WebSocket との違い​

動作フロー​

4-mode JSON デコード仕様​

オプション​

接続ライフサイクル​

制約と今後の課題​

テスト場所​