eKuiper MQTT Persistent Session: Subscriptions Not Maintained After Reconnection to EMQX Broker

[DavidwWoo]

eKuiper MQTT Persistent Session: Subscriptions Not Maintained After Reconnection to EMQX Broker

I'm running eKuiper with Docker Compose and have configured persistent session using MQTT_SOURCE__DEFAULT__ENABLECLIENTSESSION: "true". However, I'm experiencing an issue where subscriptions are not restored after reconnecting to the EMQX broker.

Docker Compose Configuration:

ekuiper:
  image: lfedge/ekuiper:v2.2.1-full
  ports:
    - "9081:9081"
    - "127.0.0.1:20498:20498"
  container_name: ekuiper
  platform: linux/arm64  # Specify ARM64 architecture
  hostname: ekuiper
  restart: unless-stopped
  user: root
  volumes:
    - /tmp/ekuiper//kuiper/data
    - /tmp/ekuiper/log:/kuiper/log
  environment:
    MQTT_SOURCE__DEFAULT__SERVER: "tcp://emqx587:1883"
    MQTT_SOURCE__DEFAULT__ENABLECLIENTSESSION: "true"
    KUIPER__BASIC__CONSOLELOG: "true"
    KUIPER__BASIC__IGNORECASE: "false"
    KUIPER__BASIC__LOGLEVEL: "debug"
    KUIPER__BASIC__DEBUG: "true"

manager:
  image: emqx/ekuiper-manager:1.9.5-plus
  container_name: ekuiper-manager
  platform: linux/arm64  # Specify ARM64 architecture
  ports:
    - "9082:9082"
  restart: unless-stopped
  environment:
    # setting default eKuiper service, works since 1.8.0
    DEFAULT_EKUIPER_ENDPOINT: "http://ekuiper:9081"

Issue Description:

Despite setting MQTT_SOURCE__DEFAULT__ENABLECLIENTSESSION: "true", I've confirmed that subscriptions are not restored when eKuiper reconnects to EMQX after a connection loss.

When checking in EMQX (5.8.7) > Monitor > Clients, I can see that other MQTT clients that successfully receive queued messages during disconnection periods maintain their subscriptions in the broker's client information even when disconnected. However, with eKuiper, the subscriptions disappear.

Questions:

1. Primary Question: Are there additional settings required to maintain subscriptions? How should I configure eKuiper so that subscriptions are maintained in EMQX even when the connection is lost?

2. Secondary Question: Is eKuiper-manager version 1.9.5-plus the latest version?

Additional Information:

1. EMQX is configured to buffer data properly. When testing with MQTTX client, I can confirm that both the session and subscriptions are maintained even after disconnecting MQTTX.

2. For eKuiper's MQTT source configuration, I've set the protocol version to 3.1.1 and QoS to 1.

Environment:

• eKuiper: v2.2.1-full
• EMQX: 5.8.7
• eKuiper-manager: 1.9.5-plus
• Platform: linux/arm64

Any guidance on resolving this persistent session subscription issue would be greatly appreciated.
Thank you :)

[ngjaying]

@Yisaer

[Yisaer]

Hi @DavidwWoo

Here is an exmaple to create a rule with mqtt source session enabled, note that the clientid must be given

1. PUT conf key in ekuiper:

PUT /metadata/sources/mqtt/confKeys/mqtt2

{
    "qos":1,
    "server": "tcp://127.0.0.1:1883",
    "protocolVersion": "3.1",
    "clientid": "id2",
    "enableClientSession": true,
}

2. create the mqtt stream with the given conf key:

POST /streams

{
    "sql":" CREATE stream  demo2 ()  WITH (DATASOURCE=\"/yisa/data\", FORMAT=\"json\", CONF_KEY=\"mqtt2\", TYPE=\"mqtt\");"
}

3. create the rule with the given stream:

POST /streams

{
  "id": "rule1",
  "sql": "select a from demo2",
  "options": {
      "sendSingle": false
  },
  "actions": [
      {
          "log":{
        
          },
          "mqtt":{
            "qos":1,
            "server": "tcp://127.0.0.1:1883",
             "topic": "/yisa/data22"
          }
      }
  ]
}

4. send the message to the topic
5. stop the rule:

POST /rules/rule1/stop

6. check the message queue for the client id to see whether the message queue changes.

7. start tuel rule and check whether the session enabled.

POST /rules/rule/start

[DavidwWoo]

I'll test this and report back
thx :)

[DavidwWoo]

curl -X GET http://localhost:9081/metadata/sources/yaml/mqtt

 "mqtt_broker1": {
    "clientid": "ekuiper1",
    "enableClientSession": true,
    "insecureSkipVerify": false,
    "password": "test1234!",
    "protocolVersion": "3.1.1",
    "qos": 1,
    "server": "tcp://emqx587:1883",
    "username": "admin"
  }

curl http://localhost:9081/streams/mqtt_stream

  {
  "Name": "mqtt_stream",
  "StreamFields": null,
  "Options": {
    "datasource": "david/test1",
    "format": "json",
    **"confKey": "mqtt_broker1"**,
    "type": "mqtt"
  },
  "StreamType": 0,
  "Statement": null
}

curl -s http://localhost:9081/rules/mqtt_b_to_kafka_b

{
  "triggered": true,
  "id": "mqtt_b_to_kafka_b",
  "sql": "select \n  ts as process_time,\n  meta(qos) as qos, \n  to_json(*) as payload\nfrom **mqtt_stream**;",
  "actions": [
    {
      "kafka": {
        "brokers": "192.168.0.113:9094",
        "bufferLength": 1024,
        "bufferPageSize": 5,
        "cleanCacheAtStop": false,
        "concurrency": 1,
        "dataTemplate": "{{.payload}}",
        "enableCache": true,
        "format": "json",
        "insecureSkipVerify": false,
        "maxAttempts": 3,
        "maxDiskCache": 10,
        "memoryCacheThreshold": 5,
        "omitIfEmpty": false,
        "requiredACKs": 1,
        "resendInterval": 2000,
        "runAsync": false,
        "saslAuthType": "none",
        "sendSingle": true,
        "topic": "david.test1"
      }
    }
  ],
  "options": {
    "debug": false,
    "isEventTime": false,
    "lateTolerance": "1s",
    "concurrency": 1,
    "bufferLength": 1024,
    "sendMetaToSink": false,
    "sendNilField": false,
    "sendError": false,
    "checkpointInterval": "5m0s",
    "restartStrategy": {
      "delay": "1s",
      "multiplier": 2,
      "maxDelay": "30s",
      "jitterFactor": 0.1
    }
  }
}

mqtt connected screen

mqtt disconnected screen

subscription is removed...

"enableClientSession" option is not working w/ my configurations.
lol