plantpulse-messaging（メッセージング層）

役割

PlantPulse の神経系として機能するメッセージブローカー層です。産業現場から入ってくるセンサーデータを収集し、モジュール間のイベントを伝達します。ブローカーは2つです — Kafka と HiveMQ(MQTT)。

項目	値
モジュール名	`plantpulse-messaging`
インストールパス	`/opt/kopens/plantpulse-platform/plantpulse-messaging/`
データ場所	`/data1/pp-data/kafka/kraft_combined_logs`
`pd` サービス	`messaging` — MASTER ノードのみ
アカウント	`PP_MQ_USER` / `PP_MQ_PASSWORD` — Kafka と HiveMQ が同じ値を共有します

STOMP(ActiveMQ)は退役しました

ブラウザのリアルタイムプッシュが SSE に変わったため、STOMP ブローカーは構成から外れました。61000 / 61004 ポートと PP_STOMP_* はもうサービスされません。古いファイアウォールルールに残っていれば整理してください。

構成

Kafka

KRaft モード(ZooKeeper なし)の単一ブローカーがコントローラーを兼ねます。

項目	値
ポート	9092 (SASL_PLAINTEXT) · 9093 (controller, 内部) · 9094 (SASL_SSL)
認証	SASL/PLAIN — 静的 JAAS(`config/jaas.conf`)。アカウント `PP_MQ_USER`
アドバタイズアドレス	`PP_KAFKA_ADVERTISED_HOST` — 起動時にホストアドレスへ誘導されます
トピック接頭辞	`PP_TOPIC_PREFIX` (デフォルト `pp`)

ブローカーが読むファイルは1つです

plantpulse-messaging/kafka/config/kafka.properties がブローカー設定で、ホストの /etc/kopens/conf/kafka.properties.template でレンダリングされます。同じディレクトリの broker.properties · server.properties · controller.properties はApache Kafka ディストリビューションが同梱するサンプルファイルであり、誰も読みません。コンテナ内で advertised.listeners=PLAINTEXT://localhost:9092 のような値を見ても驚かないでください。

テンプレートの主要行(プレースホルダーはレンダリング時に埋められます):

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

Kafka は1時間バッファです

ブローカーのデフォルト保持期間は1時間です。データの保存先は Aster TSDB · 時系列エンジン · Iceberg で、Kafka はその間のバッファです。トピックごとの保持期間と「誰がその値を決めたか」は pd retention が表示します。

主要トピック

接頭辞 pp(PP_TOPIC_PREFIX) が付きます。以下はプラットフォームコードが使用する代表的なトピックです。

トピック	用途
`event` · `event-row`	サーバー ↔ エンジンイベントバス
`pp-tag-point`	タグ値ストリーム — タグごとに「Kafka 公開」を有効にする必要があります
`pp-tag-alarm` · `pp-asset-alarm` · `pp-asset-event`	アラーム · アセットイベント
`pp-asset-data` · `pp-asset-aggregation` · `pp-asset-health-status`	アセットデータと集計
`pp-batch`	バッチジョブ
`pp-edge-status` · `pp-opc-status` · `pp-diagnostic`	エッジ · OPC · 診断ステータス
`pp-domain-changed-event`	メタデータ変更
`pp-production-oee` · `pp-production-ram` · `pp-production-ems`	生産指標

運用コマンド

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

Kafka ディストリビューションのツールを直接使うには、SASL 資格情報をクライアント設定ファイルに渡す必要があります。pd flow · pd node topic がそれを代行してくれるので、まずそちらを使ってください。

9092 が開いたからといって準備完了ではありません

ブローカーはリクエスト処理を開始する約10秒前に 9092 を開きます。そのため pd のヘルスチェックはポートではなく「アドバタイズアドレスへ ApiVersions リクエストを送り、応答を受け取れるか」を確認します。起動直後は DOWN (port 9092 is bound but … did not answer) を30秒後に再確認してください。

チューニングポイント

すべて /etc/kopens/conf/kafka.properties.template で変更し、restart-datalake.sh で反映します。

項目	デフォルト	備考
`log.retention.ms`	1時間	コンシューマーが長時間停止するサイトなら延長します。ディスクをその分使用します
`num.partitions`	4	新規トピックのパーティション数。コンシューマースレッド数と合わせます
`num.io.threads` · `num.network.threads`	8 · 3	NVMe なら io を 16 まで
`message.max.bytes`	150MB	大きなメッセージ(ファイルキュー)のために大きめに設定されています

HiveMQ (MQTT)

軽量 IoT デバイス · センサーと Edge エージェントが接続する標準 MQTT ブローカーです(Community Edition)。

項目	値
ポート	1883 (平文) · 1884 (TLS) — プロキシコンテナがホストに公開し、データレイクへ転送します
認証	username / password — セキュリティエクステンションが `conf/auth.properties` を読み取ります
アカウント	`PP_MQ_USER` / `PP_MQ_PASSWORD` (Kafka と同じ値)
設定	`mqtt/conf/config.xml` ← `/etc/kopens/conf/hivemq.xml.template`

TLS リスナーはプラットフォーム共通 keystore(/var/security/plantpulse/master/master.keystore.jks)を使用し、クライアント証明書は要求しません(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/#'

ログ

docker exec plantpulse-datalake pd logs --lines 100 messaging

ブローカー	ファイル
Kafka	`plantpulse-messaging/kafka/logs/`
HiveMQ	`plantpulse-messaging/mqtt/logs/hivemq.log` (`event.log` はメッセージ単位の監査ログなので、`pd logs` は意図的に除外します)

頻繁に発生する問題

症状	原因	対処
別のマシンのクライアントが bootstrap 後に切断される	アドバタイズアドレスがコンテナ内部アドレス	ノードファイル `PP_MASTER_IP` → 設定変更方法
`pd status messaging` が `DOWN (advertises 127.0.0.1…)`	ノードファイルにアドレスなし	同じ対処。レンダリング exit 8 と同じ原因
コンシューマー lag 増加	コンシューマーが遅い · 停止	`pd flow` の `MEMB` · `LAG`、コンシューマーコンテナのログ
ディスク満杯	保持期間を延ばしたがコンシューマー停止	`pd retention` の kafka 表、`log.retention.ms`
MQTT TLS ハンドシェイク失敗	証明書期限切れ · SAN 欠落	セキュリティ設定 — `plantpulse-certs` が再発行
ローテーション後に Kafka のみ認証失敗	`PP_MQ_PASSWORD` は両ブローカーで共通	`passwd.sh PP_MQ_PASSWORD` で両方を一緒に → パスワード

セキュリティ / 外部公開

ポート	推奨
9092 (Kafka SASL 平文)	閉域網のみ
9094 (Kafka SASL_SSL)	外部公開時に使用
1883 (MQTT 平文)	閉域網のみ
1884 (MQTT TLS)	外部公開時に使用

役割​

構成​

Kafka​

ブローカーが読むファイルは1つです​

主要トピック​

運用コマンド​

チューニングポイント​

HiveMQ (MQTT)​

ログ​

頻繁に発生する問題​

セキュリティ / 外部公開​

関連ドキュメント​

役割

構成