Apache Kafka
The gateway connects outbound as a client to topics on an external data streaming platform (Apache Kafka / Confluent Cloud / AWS MSK / Azure Event Hubs Kafka API) and performs consumer subscribe / producer send. Arriving records are stored in an in-memory cache, and the cached value is read at each tag's collection interval.
| Situation | Which mode to use |
|---|---|
| An external IT system sends values via REST POST | HTTP Push |
| An external server pushes over WebSocket | WebSocket Client |
| An external IIoT MQTT broker pushes on a topic | MQTT Client |
| Receiving a Kafka topic as a consumer | Apache Kafka (this page) |
| Reading values directly from an in-house PLC | Modbus / OPC-UA, etc. |
Registration form fields
| Field | What to enter | Example |
|---|---|---|
| IP address | Kafka broker host (bootstrap) | kafka.example.com, 10.0.0.50 |
| Port | Broker port (plaintext 9092, TLS 9093) | 9092, 9093 |
| GROUP ID | Consumer group id (optional) | plant-floor-A, edge-line1 |
| SECURITY | Security protocol | PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL |
| SASL MECH | SASL mechanism (optional) | PLAIN / SCRAM-SHA-256 / SCRAM-SHA-512 |
| SASL JAAS | SASL authentication JAAS configuration (optional) | org.apache.kafka.common.security.plain.PlainLoginModule required username="u" password="p"; |
| AUTO OFFSET | Starting point for a new group | latest (default) / earliest |
| Collection interval | Interval at which the gateway reads the cached value (ms) | 1000 |
Actual broker URL: <host>:<port> (e.g. kafka.example.com:9092). Even if you register only one bootstrap host, the broker will advertise the rest of the cluster information for you.
If GROUP ID is not specified, it is generated automatically in the form plantpulse-edge-<opc_id>.
Tag PLC address notation — 4-mode JSON
A tag's PLC address = Kafka topic + 4-mode decoder. The spec is identical to MQTT / WebSocket.
| Mode | Format | Behavior |
|---|---|---|
| SCALAR | sensors/temp or sensors/temp.value | The entire message as a String. If the message is a JSON object, falls back to raw. |
| KEY | sensors/temp:temperature | Value of a top-level JSON key (e.g. {"temperature":25.3,"humidity":60} → 25.3) |
| PATH | sensors/temp:$.data.tags.T1 | Dynamic JSON Pointer evaluation (nested keys supported) |
| RAW | sensors/temp:_raw_ | The entire value of the last record (for debugging) |
The first read call performs a lazy subscribe (returns an empty string). Cached values arrive from the next polling cycle onward.
Common use cases
| Case | How |
|---|---|
| In-house Kafka cluster | host = bootstrap broker IP, 9092 (plaintext) |
| Confluent Cloud | host = pkc-XXX.region.aws.confluent.cloud, 9092 + SASL_SSL + PLAIN mechanism + API key/secret JAAS |
| AWS MSK (IAM auth) | host = b-1.<cluster>...amazonaws.com, 9098, IAM auth (separate JAAS configuration required) |
| Azure Event Hubs (Kafka API) | host = <ns>.servicebus.windows.net, 9093 + SASL_SSL + PLAIN + Connection String JAAS |
| New consumer group + earliest | Select AUTO OFFSET=earliest to receive all existing data |
| Separating group offset management | Register with a different GROUP ID — offsets are managed per group even for the same topic |
write (producer.send)
When you write a value from the tag page or through the REST API, the producer sends it as a record to the corresponding topic (async, fire-and-forget). The String value is sent as the record value — JSON or plain text both work.
plc_address = factory/line1/cmd
value = ON
→ Kafka producer.send: topic="factory/line1/cmd" value="ON"
plc_address = factory/line1/cmd:temperature
value = 25.3
→ Kafka producer.send: topic="factory/line1/cmd" value="25.3" (콜론 뒤는 read decoder 정보, write 는 topic 만 사용)
Common problems and solutions
| Symptom | Cause | Solution |
|---|---|---|
| No values arriving | The broker is not sending records | Verify by consuming directly with kafka-console-consumer.sh --bootstrap-server <host>:9092 --topic <t> --from-beginning |
[KAFKA] connect 실패: TimeoutException | Bootstrap server unreachable | Verify telnet <host> 9092. Check whether advertised.listeners points to the external IP (broker side) |
[KAFKA] poll error: Authentication failed | SASL authentication failure | Recheck sasl-mechanism / sasl-jaas-config. Escape special characters in the password |
[KAFKA] poll error: SslHandshake | TLS truststore missing / expired | Issue the broker certificate from a trusted CA, or add it to the system truststore |
| New group receives no data | auto-offset=latest is set but there are no new records | Temporarily change to earliest, or wait until a publish occurs |