plantpulse-messaging (messaging layer)
Role
The message broker layer that serves as PlantPulse's nervous system. It collects sensor data coming from the 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 — MASTER node only |
| Credentials | PP_MQ_USER / PP_MQ_PASSWORD — Kafka and HiveMQ share the same value |
Browser real-time push changed to SSE, so the STOMP broker was removed from the image. Ports 61000 / 61004 and PP_STOMP_* are no longer serviced. Clean them up if they remain in old firewall rules.
Architecture
Kafka
Single broker in KRaft mode (no ZooKeeper) that also acts as controller.
| Item | Value |
|---|---|
| Port | 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 — defaults to host address on startup |
| Topic prefix | PP_TOPIC_PREFIX (default pp) |
The broker reads exactly one file
plantpulse-messaging/kafka/config/kafka.properties is the broker configuration, rendered from /etc/kopens/conf/kafka.properties.template on the host. The files broker.properties · server.properties · controller.properties in the same directory are example files that come with the Apache Kafka distribution and no one reads them. Don't be surprised if you see values like advertised.listeners=PLAINTEXT://localhost:9092 inside the container.
The core lines in 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's default retention is 1 hour. The data repository is Cassandra · the 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
Prefix pp (PP_TOPIC_PREFIX) is attached. Below are representative topics that platform code writes to.
| Topic | Purpose |
|---|---|
event · event-row | Server ↔ engine event bus |
pp-tag-point | Tag value stream — only filled if "Kafka publish" is enabled per tag |
pp-tag-alarm · pp-asset-alarm · pp-asset-event | Alarm · asset events |
pp-asset-data · pp-asset-aggregation · pp-asset-health-status | Asset data and aggregates |
pp-batch | Batch jobs |
pp-edge-status · pp-opc-status · pp-diagnostic | Edge · OPC · diagnostic state |
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
To use Kafka distribution tools directly, you must pass SASL credentials via a client config file. pd flow · pd node topic do that for you, so use those first.
The broker opens port 9092 approximately 10 seconds before it enables request handling. That's why the health check in pd doesn't watch the port — it sends an ApiVersions request to the advertised address and checks for a response. After startup, revisit DOWN (port 9092 is bound but … did not answer) after 30 seconds.
Tuning points
All changes are made 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 stall for a long time. Uses that much disk |
num.partitions | 4 | Partition count for new topics. Align with consumer thread count |
num.io.threads · num.network.threads | 8 · 3 | On NVMe, scale io up to 16 |
message.max.bytes | 150 MB | Sized large for big messages (file queue) |
HiveMQ (MQTT)
Standard MQTT broker for lightweight IoT devices, sensors, and Edge agents (Community Edition).
| Item | Value |
|---|---|
| Port | 1883 (plaintext) · 1884 (TLS) — proxy container publishes to host and forwards to datalake |
| Authentication | username / password — security extension reads conf/auth.properties |
| Credentials | PP_MQ_USER / PP_MQ_PASSWORD (same value as Kafka) |
| Configuration | mqtt/conf/config.xml ← /etc/kopens/conf/hivemq.xml.template |
The TLS listener uses the platform's shared 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 per-message audit log, so pd logs intentionally excludes it) |
Common issues
| Symptom | Cause | Remedy |
|---|---|---|
| Client on different box drops after bootstrap | Advertised address is internal container address | Node file PP_MASTER_IP → how to change config |
pd status messaging is DOWN (advertises 127.0.0.1…) | No address in node file | Same remedy. Same root as render exit 8 |
| Consumer lag increases | Slow or stalled consumer | pd flow in MEMB · LAG, consumer container logs |
| Disk full | Increased retention but consumer stalled | kafka table in pd retention, log.retention.ms |
| MQTT TLS handshake fails | Certificate expired · SAN missing | Security configuration — plantpulse-certs reissues |
| Auth fails on Kafka alone after rotation | PP_MQ_PASSWORD is shared by both brokers | Use passwd.sh PP_MQ_PASSWORD to update both → passwords |
Security / external exposure
| Port | Recommendation |
|---|---|
| 9092 (Kafka SASL plaintext) | Private network only |
| 9094 (Kafka SASL_SSL) | Use if exposing externally |
| 1883 (MQTT plaintext) | Private network only |
| 1884 (MQTT TLS) | Use if exposing externally |