search

bifromqio / bifromq

A Multi-Tenancy MQTT broker adopting Serverless architecture
https://bifromq.io
Apache License 2.0
614 stars 61 forks source link

docker bifromq:3.0.0版本客户端无法正常接收推送消息 #79

Closed xiaonannet closed 4 months ago

xiaonannet commented 4 months ago

Describe the bug 客户端订阅了 /v1/devices/1036150750765057/command ，接收不到消息 和协议版本无关，3.1.1和5.0都有这个问题

调用 /pub API： 入参 { "clientType": "web", "expirySeconds": "3600", "payload": "{\"head\":{\"cipherFlag\":0,\"timeStamp\":1713414003773},\"dataBody\":{\"msgType\":\"cloudReq\",\"serviceCode\":\"default_brightness_controls\",\"productIdentification\":\"5282358452289536\",\"cmd\":\"brightness_level_regulate\",\"deviceIdentification\":\"1036150750765057\",\"params\":{\"luminance\":\"111\"}},\"dataSign\":\"feeb5ca02d79dc42296760a7cf5049e3\"}", "qos": "2", "reqId": 1837295270338560, "tenantId": "1", "topic": "/v1/devices/1036150750765057/command" }

bifromq Response：<200 OK OK,OK,[content-length:"2", req_id:"1837295270338560"]>

To Reproduce

image image

客户端正常订阅了此主题

image image

Expected behavior 调用api推送消息，客户端订阅收不到

Logs

2024-04-18 02:29:28.340 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-1 2024-04-18 02:29:28.340 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-2 2024-04-18 02:29:28.340 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-2 2024-04-18 02:29:28.340 WARN [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type UNMATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 02:29:28.340 WARN [TaskQueueThread-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type UNMATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 02:29:28.340 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-4 2024-04-18 02:29:28.340 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-1 2024-04-18 02:29:28.340 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-4 2024-04-18 02:29:28.340 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-1 2024-04-18 02:33:28.501 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.Metadata [Metadata.java:408] [Producer clientId=producer-1] Resetting the last seen epoch of partition mqtt.distribution.completed.topic-0 to 0 since the associated topicId changed from null to 8-Nuv34fT3qc0aaG53KWCQ 2024-04-18 02:33:28.501 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.Metadata [Metadata.java:408] [Producer clientId=producer-1] Resetting the last seen epoch of partition mqtt.subscription.acked.topic-0 to 0 since the associated topicId changed from null to HxVZQaKFQDiie05VcZxwYw 2024-04-18 02:33:28.501 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.Metadata [Metadata.java:408] [Producer clientId=producer-1] Resetting the last seen epoch of partition mqtt.client.connected.topic-0 to 0 since the associated topicId changed from null to ad-GBqZOSB2MeBbKPVwhig 2024-04-18 02:37:28.713 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.NetworkClient [NetworkClient.java:977] [Producer clientId=producer-1] Node -1 disconnected. 2024-04-18 04:19:48.686 INFO [mqtt-worker-elg-1] --- bifromq.plugin.provider.AuthProvider [AuthProvider.java:77] Authenticating client - clientId: 1036150750765056@1, username: 123456, password: 123456 2024-04-18 04:19:48.800 INFO [pool-2-thread-2] --- bifromq.plugin.provider.AuthProvider [AuthProvider.java:103] Authentication response - statusCode: 200, responseBody: {"id":null,"createdTime":null,"createdBy":null,"updatedTime":null,"updatedBy":null,"echoMap":null,"certificationResult":true,"errorMessage":null,"deviceResult":{"id":"470808946278400017","createdTime":"2024-03-19 17:00:10","createdBy":"1452186486253289472","updatedTime":"2024-03-19 17:00:10","updatedBy":"1452186486253289472","echoMap":{},"clientId":"1036150750765056@1","userName":"123456","password":"123456","appId":"thinglinks-test","authMode":0,"encryptKey":"","encryptVector":"","signKey":"123","encryptMethod":0,"deviceIdentification":"1036150750765057","deviceName":"??????A1","connector":"broker.thinglinks.mqttsnet.com:11883","description":"","deviceStatus":1,"connectStatus":2,"deviceTags":"","productIdentification":"5282358452289536","swVersion":"v1","fwVersion":"v1","deviceSdkVersion":"v1","gatewayId":null,"nodeType":1,"remark":"","createdOrgId":null,"productResultVO":null,"deviceLocationResultVO":null},"tenantId":"1"} 2024-04-18 04:19:48.800 INFO [pool-2-thread-2] --- bifromq.plugin.provider.AuthProvider [AuthProvider.java:123] Authentication successful - clientId: 1036150750765056@1, tenantId: 1 2024-04-18 04:19:48.809 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:48.810 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-3 2024-04-18 04:19:48.810 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'CLIENT_CONNECTED'. Time: 1713413988810 2024-04-18 04:19:48.810 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: CLIENT_CONNECTED 2024-04-18 04:19:48.813 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.client.connected.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","keepAliveTimeSeconds":60,"event":"connect","timestamp":1713413988810} 2024-04-18 04:19:48.813 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: CLIENT_CONNECTED 2024-04-18 04:19:48.813 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'CLIENT_CONNECTED'. Duration: 3 ms 2024-04-18 04:19:48.813 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-3 2024-04-18 04:19:48.813 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-3 2024-04-18 04:19:48.816 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.client.connected.topic. 2024-04-18 04:19:48.816 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.client.connected.topic,end 2024-04-18 04:19:48.931 INFO [basekv-client-executor-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:48.931 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-2 2024-04-18 04:19:48.931 WARN [TaskQueueThread-2] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type MATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 04:19:48.932 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-2 2024-04-18 04:19:48.932 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-2 2024-04-18 04:19:48.934 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:48.934 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-4 2024-04-18 04:19:48.934 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'SUB_ACKED'. Time: 1713413988934 2024-04-18 04:19:48.934 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: SUB_ACKED 2024-04-18 04:19:48.936 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.subscription.acked.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","messageId":48800,"topic":"/v1/devices/1106781324582913/dataResponse","event":"SUBSCRIBE","timestamp":1713413988934} 2024-04-18 04:19:48.936 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: SUB_ACKED 2024-04-18 04:19:48.936 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'SUB_ACKED'. Duration: 2 ms 2024-04-18 04:19:48.936 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-4 2024-04-18 04:19:48.936 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-4 2024-04-18 04:19:48.938 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.subscription.acked.topic. 2024-04-18 04:19:48.938 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.subscription.acked.topic,end 2024-04-18 04:19:49.030 INFO [basekv-client-executor-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.031 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-1 2024-04-18 04:19:49.031 WARN [TaskQueueThread-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type MATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 04:19:49.031 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-1 2024-04-18 04:19:49.031 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-1 2024-04-18 04:19:49.032 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.032 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-3 2024-04-18 04:19:49.032 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'SUB_ACKED'. Time: 1713413989032 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: SUB_ACKED 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.subscription.acked.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","messageId":48801,"topic":"/v1/devices/3160832883650561/topo/addResponse","event":"SUBSCRIBE","timestamp":1713413989033} 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: SUB_ACKED 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'SUB_ACKED'. Duration: 1 ms 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-3 2024-04-18 04:19:49.033 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-3 2024-04-18 04:19:49.035 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.subscription.acked.topic. 2024-04-18 04:19:49.035 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.subscription.acked.topic,end 2024-04-18 04:19:49.130 INFO [basekv-client-executor-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.131 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-2 2024-04-18 04:19:49.131 WARN [TaskQueueThread-2] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type MATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 04:19:49.131 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-2 2024-04-18 04:19:49.131 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-2 2024-04-18 04:19:49.133 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.133 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-4 2024-04-18 04:19:49.133 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'SUB_ACKED'. Time: 1713413989133 2024-04-18 04:19:49.133 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: SUB_ACKED 2024-04-18 04:19:49.134 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.subscription.acked.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","messageId":48802,"topic":"/v1/devices/3160832883650561/topo/updateResponse","event":"SUBSCRIBE","timestamp":1713413989133} 2024-04-18 04:19:49.134 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: SUB_ACKED 2024-04-18 04:19:49.134 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'SUB_ACKED'. Duration: 1 ms 2024-04-18 04:19:49.134 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-4 2024-04-18 04:19:49.134 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-4 2024-04-18 04:19:49.136 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.subscription.acked.topic. 2024-04-18 04:19:49.136 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.subscription.acked.topic,end 2024-04-18 04:19:49.232 INFO [basekv-client-executor-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.233 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-1 2024-04-18 04:19:49.233 WARN [TaskQueueThread-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type MATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 04:19:49.233 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-1 2024-04-18 04:19:49.233 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-1 2024-04-18 04:19:49.235 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.235 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-3 2024-04-18 04:19:49.235 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'SUB_ACKED'. Time: 1713413989235 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: SUB_ACKED 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.subscription.acked.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","messageId":48803,"topic":"/v1/devices/3160832883650561/topo/queryResponse","event":"SUBSCRIBE","timestamp":1713413989236} 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: SUB_ACKED 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'SUB_ACKED'. Duration: 1 ms 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-3 2024-04-18 04:19:49.236 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-3 2024-04-18 04:19:49.244 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.subscription.acked.topic. 2024-04-18 04:19:49.244 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.subscription.acked.topic,end 2024-04-18 04:19:49.336 INFO [basekv-client-executor-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.336 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-2 2024-04-18 04:19:49.336 WARN [TaskQueueThread-2] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type MATCHED as no mapping exists in TOPIC_MAP. 2024-04-18 04:19:49.336 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-2 2024-04-18 04:19:49.336 INFO [TaskQueueThread-2] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-2 2024-04-18 04:19:49.338 INFO [mqtt-worker-elg-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-4 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'SUB_ACKED'. Time: 1713413989338 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: SUB_ACKED 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.subscription.acked.topic: {"clientId":"1036150750765056@1","address":"/111.202.125.68:11173","success":"success","tenantId":"1","messageId":48804,"topic":"/v1/devices/1036150750765057/command","event":"SUBSCRIBE","timestamp":1713413989338} 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: SUB_ACKED 2024-04-18 04:19:49.338 INFO [TaskQueueThread-4] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'SUB_ACKED'. Duration: 0 ms 2024-04-18 04:19:49.339 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-4 2024-04-18 04:19:49.339 INFO [TaskQueueThread-4] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-4 2024-04-18 04:19:49.341 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.subscription.acked.topic. 2024-04-18 04:19:49.341 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.subscription.acked.topic,end 2024-04-18 04:20:04.137 INFO [topic-matcher-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:20:04.137 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-1 2024-04-18 04:20:04.137 WARN [TaskQueueThread-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:101] Discarding events of type DELIVERED as no mapping exists in TOPIC_MAP. 2024-04-18 04:20:04.137 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-1 2024-04-18 04:20:04.137 INFO [TaskQueueThread-1] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-1 2024-04-18 04:20:04.138 INFO [basekv-client-executor] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:67] Adding new task... 2024-04-18 04:20:04.138 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:48] Beginning task execution on thread: TaskQueueThread-3 2024-04-18 04:20:04.138 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:140] Starting execution of event: 'DISTED'. Time: 1713414004138 2024-04-18 04:20:04.138 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:151] Processing event type: DISTED 2024-04-18 04:20:04.140 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:159] Sent to Kafka topic mqtt.distribution.completed.topic: {"expiryInterval":3600,"qos":2,"payload":"{\"head\":{\"cipherFlag\":0,\"timeStamp\":1713414003773},\"dataBody\":{\"msgType\":\"cloudReq\",\"serviceCode\":\"default_brightness_controls\",\"productIdentification\":\"5282358452289536\",\"cmd\":\"brightness_level_regulate\",\"deviceIdentification\":\"1036150750765057\",\"params\":{\"luminance\":\"111\"}},\"dataSign\":\"feeb5ca02d79dc42296760a7cf5049e3\"}","tenantId":"1","topic":"/v1/devices/1036150750765057/command","messageId":0,"time":0,"event":"PUBLISH","body":"{\"head\":{\"cipherFlag\":0,\"timeStamp\":1713414003773},\"dataBody\":{\"msgType\":\"cloudReq\",\"serviceCode\":\"default_brightness_controls\",\"productIdentification\":\"5282358452289536\",\"cmd\":\"brightness_level_regulate\",\"deviceIdentification\":\"1036150750765057\",\"params\":{\"luminance\":\"111\"}},\"dataSign\":\"feeb5ca02d79dc42296760a7cf5049e3\"}","timestamp":0} 2024-04-18 04:20:04.140 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:164] Processed event type: DISTED 2024-04-18 04:20:04.140 INFO [TaskQueueThread-3] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:147] Completed execution of event: 'DISTED'. Duration: 2 ms 2024-04-18 04:20:04.140 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:50] Task execution completed on thread: TaskQueueThread-3 2024-04-18 04:20:04.140 INFO [TaskQueueThread-3] --- bifromq.plugin.utils.TaskQueue [TaskQueue.java:46] Waiting for task, executing on thread: TaskQueueThread-3 2024-04-18 04:20:04.147 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:436] Message successfully sent to topic mqtt.distribution.completed.topic. 2024-04-18 04:20:04.147 INFO [kafka-producer-network-thread | producer-1] --- b.plugin.provider.EventKafkaProvider [EventKafkaProvider.java:438] topic:mqtt.distribution.completed.topic,end 2024-04-18 04:22:29.259 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.Metadata [Metadata.java:408] [Producer clientId=producer-1] Resetting the last seen epoch of partition mqtt.subscription.acked.topic-0 to 0 since the associated topicId changed from null to HxVZQaKFQDiie05VcZxwYw 2024-04-18 04:22:29.260 INFO [kafka-producer-network-thread | producer-1] --- org.apache.kafka.clients.Metadata [Metadata.java:408] [Producer clientId=producer-1] Resetting the last seen epoch of partition mqtt.client.connected.topic-0 to 0 since the associated topicId changed from null to ad-GBqZOSB2MeBbKPVwhig

HOST

OS(please complete the following information):

JVM:

BifroMQ

image

插件是基于3.0.0重新编译的，代码是bifromq2.0正常使用的，所以和插件无关

Additional context

image

info： 2024-04-18 02:28:29.997 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: distservice.DistService 2024-04-18 02:28:30.204 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: inboxservice.InboxService 2024-04-18 02:28:30.405 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: sessiondict.SessionDictService 2024-04-18 02:28:30.606 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: mqttbroker.OnlineInboxBroker 2024-04-18 02:28:30.808 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: retainservice.RetainService 2024-04-18 02:28:31.010 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: dist.worker@basekv.BaseKVStoreService 2024-04-18 02:28:31.211 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: retain.store@basekv.BaseKVStoreService 2024-04-18 02:28:31.412 INFO [main] --- com.baidu.bifromq.baserpc.RPCServer [RPCServer.java:179] Start server register for service: inbox.store@basekv.BaseKVStoreService 2024-04-18 02:28:31.616 INFO [main] --- c.baidu.bifromq.mqtt.AbstractMQTTBroker [AbstractMQTTBroker.java:98] Starting MQTT broker 2024-04-18 02:28:31.618 INFO [main] --- c.baidu.bifromq.mqtt.AbstractMQTTBroker [AbstractMQTTBroker.java:121] MQTT broker started 2024-04-18 02:28:31.618 INFO [main] --- com.baidu.bifromq.apiserver.APIServer [APIServer.java:108] Starting API server 2024-04-18 02:28:31.618 INFO [main] --- com.baidu.bifromq.apiserver.APIServer [APIServer.java:115] API server started 2024-04-18 02:28:36.429 INFO [main] --- c.b.bifromq.starter.StandaloneStarter [StandaloneStarter.java:557] Standalone broker started

gc： [2024-04-18T02:28:26.762+0000][64][gc] Using The Z Garbage Collector [2024-04-18T02:28:28.765+0000][69][gc] GC(0) Garbage Collection (Warmup) 234M(11%)->84M(4%) [2024-04-18T02:32:26.032+0000][69][gc] GC(1) Garbage Collection (Warmup) 410M(20%)->112M(5%) [2024-04-18T03:01:17.606+0000][69][gc] GC(2) Garbage Collection (Warmup) 616M(30%)->112M(5%) [2024-04-18T03:06:17.789+0000][69][gc] GC(3) Garbage Collection (Proactive) 202M(10%)->106M(5%) [2024-04-18T03:11:17.963+0000][69][gc] GC(4) Garbage Collection (Proactive) 204M(10%)->108M(5%) [2024-04-18T03:16:18.066+0000][69][gc] GC(5) Garbage Collection (Proactive) 202M(10%)->104M(5%) [2024-04-18T03:21:18.163+0000][69][gc] GC(6) Garbage Collection (Proactive) 210M(10%)->100M(5%) [2024-04-18T03:26:18.260+0000][69][gc] GC(7) Garbage Collection (Proactive) 206M(10%)->102M(5%) [2024-04-18T03:31:18.357+0000][69][gc] GC(8) Garbage Collection (Proactive) 208M(10%)->104M(5%) [2024-04-18T03:36:18.458+0000][69][gc] GC(9) Garbage Collection (Proactive) 208M(10%)->106M(5%) [2024-04-18T03:41:18.556+0000][69][gc] GC(10) Garbage Collection (Proactive) 210M(10%)->106M(5%) [2024-04-18T03:46:18.657+0000][69][gc] GC(11) Garbage Collection (Proactive) 208M(10%)->104M(5%) [2024-04-18T03:51:18.761+0000][69][gc] GC(12) Garbage Collection (Proactive) 212M(10%)->102M(5%) [2024-04-18T03:56:18.862+0000][69][gc] GC(13) Garbage Collection (Proactive) 212M(10%)->100M(5%) [2024-04-18T04:01:18.958+0000][69][gc] GC(14) Garbage Collection (Proactive) 216M(11%)->106M(5%) [2024-04-18T04:06:19.060+0000][69][gc] GC(15) Garbage Collection (Proactive) 212M(10%)->100M(5%) [2024-04-18T04:11:19.157+0000][69][gc] GC(16) Garbage Collection (Proactive) 218M(11%)->102M(5%) [2024-04-18T04:16:19.263+0000][69][gc] GC(17) Garbage Collection (Proactive) 216M(11%)->100M(5%) [2024-04-18T04:21:19.360+0000][69][gc] GC(18) Garbage Collection (Proactive) 228M(11%)->102M(5%) [2024-04-18T04:26:19.468+0000][69][gc] GC(19) Garbage Collection (Proactive) 220M(11%)->108M(5%) [2024-04-18T04:31:19.558+0000][69][gc] GC(20) Garbage Collection (Proactive) 222M(11%)->102M(5%) [2024-04-18T04:36:19.667+0000][69][gc] GC(21) Garbage Collection (Proactive) 224M(11%)->106M(5%) [2024-04-18T04:41:19.759+0000][69][gc] GC(22) Garbage Collection (Proactive) 226M(11%)->110M(5%)

popduke commented 4 months ago

已确认是通过HTTP API向MQTT 5.0 client pub指定了可选参数expiry_seconds的消息的bug，不指定该参数功能正常。

popduke commented 4 months ago

已在3.0.2中修复