plantpulse-messaging (Messaging Layer)
Role
This is the message broker layer that serves as the nervous system of PlantPulse. It collects sensor data coming from the industrial field and delivers events between modules. There are two brokers — Kafka and HiveMQ (MQTT).
| Item | Value |
|---|---|
| Module Name | plantpulse-messaging |
| Installation Path | /opt/kopens/plantpulse-platform/plantpulse-messaging/ |
| Data Location | /data1/pp-data/kafka/kraft_combined_logs |
pd Service | messaging — only on the MASTER node |
| Credentials | PP_MQ_USER / PP_MQ_PASSWORD — Kafka and HiveMQ share the same values |
With browser real-time push moving to SSE, the STOMP broker has been removed from the image. Ports 61000 / 61004 and PP_STOMP_* are no longer served. Clean them up if they remain in old firewall rules.
Configuration
Kafka
A single broker in KRaft mode (no ZooKeeper) also serves as the controller.
| Item | Value |
|---|---|
| Ports | 9092 (SASL_PLAINTEXT) · 9093 (controller, internal) · 9094 (SASL_SSL) |
| Authentication | SASL/PLAIN — static JAAS(config/jaas.conf). Credentials PP_MQ_USER |
| Advertised Address | PP_KAFKA_ADVERTISED_HOST — derived from the host address at startup |
| Topic Prefix | PP_TOPIC_PREFIX (default pp) |
There is only one file the broker reads
plantpulse-messaging/kafka/config/kafka.properties is the broker configuration, rendered from /etc/kopens/conf/kafka.properties.template on the host. The broker.properties · server.properties · controller.properties in the same directory are example files shipped with the Apache Kafka distribution and are read by no one. Don't be surprised to see values like advertised.listeners=PLAINTEXT://localhost:9092 inside the container.
The core lines of the template (placeholders are filled at render time):
process.roles=broker,controller
listeners=SASL_PLAINTEXT://0.0.0.0:${PP_KAFKA_PORT},CONTROLLER://0.0.0.0:${PP_KAFKA_CONTROLLER_PORT:9093},SASL_SSL://0.0.0.0:${PP_KAFKA_TLS_PORT}
advertised.listeners=SASL_PLAINTEXT://${PP_KAFKA_ADVERTISED_HOST}:${PP_KAFKA_PORT},…
sasl.enabled.mechanisms=PLAIN
allow.everyone.if.no.acl.found=false
log.dirs=${PP_DATA_DIR}/kafka/kraft_combined_logs
log.retention.ms=3600000 # 1시간 — 버퍼이지 보관소가 아닙니다
log.segment.bytes=10485760 # 10MB
num.partitions=4
default.replication.factor=1 # 단일 브로커
message.max.bytes=157286400
auto.create.topics.enable=true
The broker default retention is 1 hour. The data store is Aster TSDB · Time Series Engine · Iceberg, and Kafka is the buffer in between. Per-topic retention and "who set that value" are shown by pd retention.
Key Topics
They are prefixed with pp(PP_TOPIC_PREFIX). Below are the representative topics used by platform code.
| Topic | Purpose |
|---|---|
event · event-row | Server ↔ Engine event bus |
pp-tag-point | Tag value stream — it only fills in if "Kafka publish" is enabled per tag |
pp-tag-alarm · pp-asset-alarm · pp-asset-event | Alarms · Asset events |
pp-asset-data · pp-asset-aggregation · pp-asset-health-status | Asset data and aggregations |
pp-batch | Batch jobs |
pp-edge-status · pp-opc-status · pp-diagnostic | Edge · OPC · diagnostics status |
pp-domain-changed-event | Metadata changes |
pp-production-oee · pp-production-ram · pp-production-ems | Production metrics |
Operational Commands
docker exec plantpulse-datalake pd status messaging # UP (advertised <address>:9092 answered ApiVersions)
docker exec plantpulse-datalake pd flow # 컨슈머 그룹별 lag
docker exec plantpulse-datalake pd node topic # 토픽 "event" describe (비밀번호 자동 주입)
docker exec plantpulse-datalake pd retention # 토픽별 보존과 세그먼트 크기
docker exec plantpulse-datalake pd restart messaging
If you want to use the tools from the Kafka distribution directly, you must pass SASL credentials via a client configuration file. pd flow · pd node topic handle that for you, so use them first.
The broker opens 9092 about 10 seconds before it starts processing requests. That's why the pd health check doesn't look at the port — it checks whether an ApiVersions request sent to the advertised address gets a response. If DOWN (port 9092 is bound but … did not answer) fails right after startup, check again after 30 seconds.
Tuning Points
All of these are changed in /etc/kopens/conf/kafka.properties.template and applied via restart-datalake.sh.
| Item | Default | Notes |
|---|---|---|
log.retention.ms | 1 hour | Increase if consumers can be stopped for long periods. It consumes disk space accordingly |
num.partitions | 4 | Number of partitions for new topics. Matches the number of consumer threads |
num.io.threads · num.network.threads | 8 · 3 | If NVMe, you can raise io to 16 |
message.max.bytes | 150MB | Set high for large messages (file queues) |
HiveMQ (MQTT)
Standard MQTT broker for lightweight IoT devices · sensors and Edge agents (Community Edition).
| Item | Value |
|---|---|
| Ports | 1883 (plaintext) · 1884 (TLS) — The proxy container publishes to the host and forwards to the data lake |
| Authentication | username / password — the security extension reads conf/auth.properties |
| Credentials | PP_MQ_USER / PP_MQ_PASSWORD (same values as Kafka) |
| Configuration | mqtt/conf/config.xml ← /etc/kopens/conf/hivemq.xml.template |
The TLS listener uses the platform common keystore (/var/security/plantpulse/master/master.keystore.jks) and does not require client certificates (NONE).
# 외부에서 발행 · 구독 (디버깅)
mosquitto_pub -h <server-ip> -p 1883 -u mq -P '<PP_MQ_PASSWORD>' -t test/topic -m "hello"
mosquitto_sub -h <server-ip> -p 1883 -u mq -P '<PP_MQ_PASSWORD>' -t 'test/#'
Logs
docker exec plantpulse-datalake pd logs --lines 100 messaging
| Broker | File |
|---|---|
| Kafka | plantpulse-messaging/kafka/logs/ |
| HiveMQ | plantpulse-messaging/mqtt/logs/hivemq.log (event.log is a message-level audit log, so pd logs intentionally excludes it) |
Common Issues
| Symptom | Cause | Action |
|---|---|---|
| Clients on another machine disconnect after bootstrap | Advertised address is the container's internal address | Node file PP_MASTER_IP → How to change settings |
pd status messaging is DOWN (advertises 127.0.0.1…) | No address in the node file | Same action. Same cause as render exit 8 |
| Consumer lag increases | Consumer is slow · stopped | pd flow of MEMB · LAG, consumer container logs |
| Disk full | Increased retention but consumer stopped | Kafka table in pd retention, log.retention.ms |
| MQTT TLS handshake fails | Certificate expired · SAN missing | Security settings — plantpulse-certs reissues |
| Only Kafka authentication fails after rotation | PP_MQ_PASSWORD is shared by both brokers | Use passwd.sh PP_MQ_PASSWORD to update both together → Passwords |
Security / External Exposure
| Port | Recommendation |
|---|---|
| 9092 (Kafka SASL plaintext) | Private network only |
| 9094 (Kafka SASL_SSL) | Use for external exposure |
| 1883 (MQTT plaintext) | Private network only |
| 1884 (MQTT TLS) | Use for external exposure |