Apache Kafka Treiber

Der Apache Kafka Treiber von PlantPulse Edge ist eine Java-Implementierung, die die kafka-clients 4.1.2 Bibliothek kapselt. Er abonniert als Consumer Topics einer externen Streaming-Plattform (Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API), hält die Records in einem In-Memory-Cache und gibt den Cache-Wert im Polling-Zyklus jedes Tags zurück. Schreiben erfolgt über producer.send (asynchron).

Quelle: plantpulse.driver.protocol.kafka.*

Klasse	Verantwortung
`KafkaDriver`	`ProtocolDriver` Implementierung — connect/read/write/close, Topic-Cache, lazy subscribe, Hintergrund-Poller-Thread

Bibliothek: lib/kafka-clients-4.1.2.jar

Unterschied zu MQTT / WebSocket

PlantPulse bietet drei message-basierte Treiber — alle gemeinsam nutzen denselben JsonExtract 4-Modus-JSON-Decoder.

Treiber	Wire-Protokoll	Payload	Einsatzbereich
`MQTT_CLIENT`	MQTT v3.1.1	beliebig (meist JSON/Klartext)	IIoT-Broker (HiveMQ / Mosquitto / EMQX), QoS 0/1/2
`WEBSOCKET`	ws/wss	beliebig (meist JSON)	Streaming-Server / proprietäre Push-API
`KAFKA` (diese Seite)	Kafka-Wire-Protokoll	Record-Key/Value (meist JSON-String)	High-Throughput-Streaming, Replay (Offset), Consumer-Group
`SPARKPLUG_B`	MQTT + Protobuf	Sparkplug-B-Payload	Cirrus Link / Tahu / Ignition Standard (IIoT-Spec über MQTT)

Bei Kafka stehen persistentes Log + Offset-Replay im Mittelpunkt — ein neuer Consumer, der sich mit auto-offset=earliest registriert, kann Daten ab dem letzten Stand erhalten (innerhalb der Retention des Brokers).

Ablauf

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

Beim ersten read-Aufruf ergibt ein Cache-Miss ein lazy subscribe (Aufruf consumer.subscribe(...)) und Rückgabe eines leeren Strings — ab dem nächsten Zyklus liegen echte Werte vor. Topics, die in tag_map registriert sind, werden direkt nach dem Verbindungsaufbau über preSubscribeFromTagMap() gesammelt abonniert, sodass Werte bereits im ersten Zyklus ankommen können.

4-Modus-JSON-Dekodierspezifikation

Das JsonExtract-Hilfsprogramm vereinheitlicht die Nachrichtendekodierung der drei Treiber MQTT / WebSocket / Kafka.

Modus	Adressformat	Verhalten
SCALAR	`<topic>` oder `<topic>.value`	gesamte Nachricht als String. Bei JSON-Objekt: Rohdaten-Fallback.
KEY	`<topic>:<json-key>`	Wert eines Schlüssels im Top-Level-JSON-Objekt
PATH	`<topic>:$.<json-path>`	dynamische Auswertung per JSON Pointer (z. B. `$.data.tags.T1` → `/data/tags/T1`)
RAW	`<topic>:_raw_`	letzte Rohdaten-Nachricht (Debugging)

Cache-Struktur — verwendet einen einzigen Map<String, String>, Schlüssel ist topic + "|" + field:

Schlüssel	Bedeutung
`topic\|_raw_`	letzte Rohdaten-Nachricht
`topic\|value`	Skalar (nur bei nicht-JSON-Nachricht)
`topic\|<json-key>`	Werte je Top-Level-JSON-Schlüssel (nur bei Objekt)

Der PATH-Modus umgeht den Cache und wertet die letzte Rohdaten-Nachricht aus rawByTopic bei jedem Aufruf mit JSONPointer.queryFrom(...) aus.

Optionen

Schlüssel	Standard	Beschreibung
`group-id`	`plantpulse-edge-<opc_id>`	Consumer-Gruppen-ID. Einheit für Offset-Commit.
`security-protocol`	`PLAINTEXT`	`PLAINTEXT` / `SSL` / `SASL_PLAINTEXT` / `SASL_SSL`
`sasl-mechanism`	(keine)	`PLAIN` / `SCRAM-SHA-256` / `SCRAM-SHA-512`
`sasl-jaas-config`	(keine)	JAAS-Konfiguration (z. B. `...PlainLoginModule required username="u" password="p";`)
`auto-offset`	`latest`	`earliest` / `latest` — Startpunkt einer neuen Gruppe

enable.auto.commit=true ist stets gesetzt, sodass der Broker Offsets automatisch committet (Standard: 5-Sekunden-Intervall).

Verbindungslebenszyklus

Methode	Aktion
`connect()`	Optionen anwenden → Consumer/Producer erzeugen → Poller-Thread starten → `tag_map` gesammelt abonnieren
`pollerLoop()`	`consumer.poll(500ms)` wiederholen → je Record `JsonExtract.updateCache(...)`
`read()`	Parsen → bei Cache-Miss lazy subscribe → bei Cache-Hit Wert zurückgeben
`write()`	`producer.send(new ProducerRecord<>(topic, value))` (asynchron)
`close()`	`consumer.wakeup()` → Thread-Join → Consumer/Producer schließen → Cache leeren

Grenzen und zukünftige Arbeiten

Wiederverbindung — kafka-clients besitzt eine eigene Reconnect-Logik; wenn jedoch der gesamte Broker ausfällt und langjährige poll() fehlschlagen, markiert der Treiber setConnected(false) nicht separat → für Betriebsstatus-Monitoring werden Broker-seitige Metriken empfohlen.
AWS MSK IAM — aws-msk-iam-auth-JAR erforderlich (derzeit nicht enthalten). Später aws-msk-iam-auth-1.x.jar in lib/ aufnehmen und sasl.client.callback.handler.class als Option freigeben.
TLS-Truststore — nicht über Treiberoptionen konfigurierbar; es wird der JVM-System-Truststore verwendet. Bei einem Broker mit selbstsignierter privater CA das CA-Zertifikat in den JVM-Truststore aufnehmen oder der Gateway-Startoption -Djavax.net.ssl.trustStore=… hinzufügen.
Producer mit partition/key-Angabe — Write erfolgt immer ohne Key, sodass die Partition per Round-Robin bestimmt wird. Das bedeutet: Schreibvorgänge desselben Tags kommen nicht garantiert in Reihenfolge an. Ist eine partitionäre Reihenfolge erforderlich, das Topic mit einer Partition anlegen oder die Reihenfolge auf Consumer-Seite sicherstellen.
Transaktionaler Producer — nicht unterstützt. Ein idempotenter Producer wird derzeit ebenfalls nicht explizit aktiviert.

Testorte

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

Klassenladung / Hierarchie (BaseProtocolDriver / ProtocolDriver)
Anfangszustand / Metadaten (isConnected/isWriteSupported/isExternal/getDriverSource)
Nicht verbundener Zugriff: read/write/null sicher
getDebugBaseUrl — kafka://, Standard-Port (9092)
Optionsparsing — group-id / security-protocol / sasl-* / auto-offset, options=null sicher
4-Modus-Dekodierung — Cache direkt befüllt, KEY / PATH / RAW / SCALAR / .value-Suffix werden alle verifiziert
close() nicht verbunden sicher

test/java/plantpulse/driver/protocol/common/JsonExtractTest.java — 23 Tests (integrierte Verifikation von 4-Modus-Parser/Cache/Extraktion).

Integrationstests mit echtem Broker sind separat — diese Unit-Tests laufen ohne externes Kafka.

Unterschied zu MQTT / WebSocket​

Ablauf​

4-Modus-JSON-Dekodierspezifikation​

Optionen​

Verbindungslebenszyklus​

Grenzen und zukünftige Arbeiten​

Testorte​