Apache Kafka Driver
The Apache Kafka driver for PlantPulse Edge is a Java implementation that wraps 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),
stores records in an in-memory cache, and returns the cached value on each tag polling cycle.
Writes are done via 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 share the same JsonExtract 4-mode JSON decoder.
| Driver | Wire Protocol | Payload | When to Use |
|---|---|---|---|
MQTT_CLIENT | MQTT v3.1.1 | Arbitrary (usually JSON/plain text) | IIoT broker (HiveMQ / Mosquitto / EMQX), QoS 0/1/2 |
WEBSOCKET | ws/wss | Arbitrary (usually JSON) | streaming server / proprietary push API |
KAFKA (this page) | Kafka wire protocol | record key/value (usually 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) |
Kafka's core feature is persistent log + offset replay — when a new consumer registers with auto-offset=earliest,
it can receive data from the past (within the broker's retention).
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
On the first read call, if there's a cache miss, a lazy subscribe occurs (consumer.subscribe(...) call) and an empty string is returned —
actual values arrive from the next cycle onward. Topics registered in tag_map are batch-subscribed via
preSubscribeFromTagMap() right after connect, so values may arrive on the very first cycle.
4-mode JSON Decoding Spec
The JsonExtract utility unifies message decoding across the MQTT / WebSocket / Kafka drivers.
| Mode | Address Format | Behavior |
|---|---|---|
| SCALAR | <topic> or <topic>.value | Entire message as a String. Raw fallback if it's a JSON object. |
| KEY | <topic>:<json-key> | Value of one key in a 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, keyed by topic + "|" + field:
| Key | Meaning |
|---|---|
topic|_raw_ | Last raw message |
topic|value | Scalar (only when the message isn't JSON) |
topic|<json-key> | Value per top-level JSON key (only when it's an object) |
PATH mode doesn't use the cache — it evaluates the last raw message from rawByTopic each time using JSONPointer.queryFrom(...).
Options
| Key | Default | Description |
|---|---|---|
group-id | plantpulse-edge-<opc_id> | Consumer group ID. Unit for 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 automatically commits offsets (default 5-second interval).
Connection Lifecycle
| Method | Behavior |
|---|---|
connect() | Apply options → create consumer / producer → start poller thread → batch subscribe via tag_map |
pollerLoop() | Loop over consumer.poll(500ms) → per-record JsonExtract.updateCache(...) |
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 marksetConnected(false)→ monitoring operational status via broker-side metrics is recommended. - AWS MSK IAM — the
aws-msk-iam-authjar needs to be added (not currently included). In the future, we plan to placeaws-msk-iam-auth-1.x.jarinlib/and exposesasl.client.callback.handler.classas an option. - TLS truststore — cannot be specified as a driver option; the JVM system truststore is used. If your broker uses a self-signed private CA, add the CA certificate to the JVM's truststore, or add
-Djavax.net.ssl.trustStore=…to the gateway startup options. - Partition / key-specified producer — writes are always sent without a key, so partitions are determined by round-robin. This means there is no guarantee that writes for the same tag arrive in order. If per-partition ordering is required, create the topic with one partition or handle ordering on the consumer side.
- Transactional producer — not supported. Currently, the idempotent producer is also not explicitly enabled.
Test Locations
test/java/plantpulse/driver/protocol/kafka/KafkaDriverTest.java — 18 tests.
- Class loading / hierarchy (
BaseProtocolDriver/ProtocolDriver) - Initial state / metadata (
isConnected/isWriteSupported/isExternal/getDriverSource) - Disconnected read/write/null safety
getDebugBaseUrl—kafka://, default port (9092)- Option parsing —
group-id/security-protocol/sasl-*/auto-offset, options=null safety - 4-mode decoding — direct cache population to verify KEY / PATH / RAW / SCALAR /
.valuesuffix close()disconnected safety
test/java/plantpulse/driver/protocol/common/JsonExtractTest.java — 23 tests (integrated verification of 4-mode parser/cache/extract).
Integration tests that depend on a real broker are separate — these unit tests operate without an external Kafka.