Apache Kafka Driver — Technical Reference

The Apache Kafka driver in PlantPulse Edge is a Java implementation wrapping the kafka-clients 4.1.2 library. It subscribes as a consumer to topics on an external streaming platform (Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API), holds the records in an in-memory cache, and returns the cached value on every tag polling cycle. Writes go through producer.send (async).

Source: plantpulse.driver.protocol.kafka.*

Class	Responsibility
`KafkaDriver`	`ProtocolDriver` implementation — connect/read/write/close, topic cache, lazy subscribe, background poller thread

Library: lib/kafka-clients-4.1.2.jar

Differences from MQTT / WebSocket

PlantPulse provides three message-based drivers — all of which share the same JsonExtract 4-mode JSON decoder.

Driver	Wire protocol	Payload	When to use
`MQTT_CLIENT`	MQTT v3.1.1	Arbitrary (typically JSON/plain text)	IIoT broker (HiveMQ / Mosquitto / EMQX), QoS 0/1/2
`WEBSOCKET`	ws/wss	Arbitrary (typically JSON)	Streaming server / proprietary push API
`KAFKA` (this page)	Kafka wire protocol	Record key/value (typically JSON String)	High-throughput streaming, replay (offset), consumer group
`SPARKPLUG_B`	MQTT + Protobuf	Sparkplug B Payload	Cirrus Link / Tahu / Ignition standard (IIoT spec over MQTT)

The core of Kafka is its persistent log + offset replay — a new consumer registering with auto-offset=earliest can receive past data as well (within the broker's retention window).

Operation Flow

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

If the first read call results in a cache miss, the driver performs a lazy subscribe (calling consumer.subscribe(...)) and returns an empty string — actual values arrive from the next cycle onward. Topics registered in tag_map are subscribed in bulk via preSubscribeFromTagMap() immediately after connect, so values may already arrive on the first cycle.

4-Mode JSON Decoding Spec

The JsonExtract utility unifies message decoding across the MQTT, WebSocket, and Kafka drivers.

Mode	Address format	Behavior
SCALAR	`<topic>` or `<topic>.value`	Entire message as a String. Falls back to raw if it is a JSON object.
KEY	`<topic>:<json-key>`	Value of one key in the top-level JSON object
PATH	`<topic>:$.<json-path>`	Dynamic JSON Pointer evaluation (e.g. `$.data.tags.T1` → `/data/tags/T1`)
RAW	`<topic>:_raw_`	Last raw message (for debugging)

Cache structure — a single Map<String, String> is used, with keys of the form topic + "|" + field:

Key	Meaning
`topic\|_raw_`	Last raw message
`topic\|value`	scalar (only when the message is not JSON)
`topic\|<json-key>`	Value per top-level JSON key (only when it is an object)

PATH mode bypasses the cache and evaluates the last raw message from rawByTopic with JSONPointer.queryFrom(...) on every call.

Options

Key	Default	Description
`group-id`	`plantpulse-edge-<opc_id>`	consumer group id. Unit of offset commit.
`security-protocol`	`PLAINTEXT`	`PLAINTEXT` / `SSL` / `SASL_PLAINTEXT` / `SASL_SSL`
`sasl-mechanism`	(none)	`PLAIN` / `SCRAM-SHA-256` / `SCRAM-SHA-512`
`sasl-jaas-config`	(none)	JAAS configuration (e.g. `...PlainLoginModule required username="u" password="p";`)
`auto-offset`	`latest`	`earliest` / `latest` — starting point for a new group

enable.auto.commit=true is always set, so the broker commits offsets automatically (every 5 seconds by default).

Connection Lifecycle

Method	Behavior
`connect()`	Apply options → create consumer / producer → start poller thread → bulk subscribe `tag_map`
`pollerLoop()`	Repeat `consumer.poll(500ms)` → `JsonExtract.updateCache(...)` per record
`read()`	Parse → lazy subscribe on cache miss → return value on cache hit
`write()`	`producer.send(new ProducerRecord<>(topic, value))` (async)
`close()`	`consumer.wakeup()` → thread join → close consumer/producer → clear cache

Limitations and Future Work

Reconnection — kafka-clients has its own reconnection logic, but when the entire broker goes down and a long-lived poll() fails, the driver does not separately mark setConnected(false) → broker-side metrics are recommended for operational status monitoring.
AWS MSK IAM — requires adding the aws-msk-iam-auth jar (not currently bundled). Placing aws-msk-iam-auth-1.x.jar in lib/ and exposing sasl.client.callback.handler.class as an option is under consideration.
TLS truststore — cannot be specified through driver options; the JVM system truststore is used. For a broker using a self-signed private CA, add the CA certificate to the JVM truststore, or add -Djavax.net.ssl.trustStore=… to the gateway startup options.
Producer with partition / key — writes are always sent without a key, so the partition is chosen round-robin. That is, there is no guarantee that writes for the same tag arrive in order. If per-partition ordering is required, create the topic with a single partition, or handle ordering on the consumer side.
Transactional producer — not supported. The idempotent producer is currently not explicitly enabled either.

Test Locations

test/java/plantpulse/driver/protocol/kafka/KafkaDriverTest.java — 18 tests.

Class loading / hierarchy (BaseProtocolDriver / ProtocolDriver)
Initial state / metadata (isConnected/isWriteSupported/isExternal/getDriverSource)
Unconnected read/write/null safety
getDebugBaseUrl — kafka://, default port (9092)
Option parsing — group-id / security-protocol / sasl-* / auto-offset, options=null safety
4-mode decoding — populating the cache directly to verify KEY / PATH / RAW / SCALAR and the .value suffix
close() unconnected safety

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java — 23 tests (integrated verification of the 4-mode parser/cache/extract).

Integration tests that depend on a real broker are maintained separately — these unit tests run without an external Kafka.

Differences from MQTT / WebSocket​

Operation Flow​

4-Mode JSON Decoding Spec​

Options​

Connection Lifecycle​

Limitations and Future Work​

Test Locations​