MQTT 5 messages sent when Ditto was not connected are not processed

[dimabarbul]

Messages that are sent to MQTT 5 broker (I used mosquitto) are delivered to Ditto, but are not processed.

Setup:

• running Ditto
• running MQTT 5 broker, e.g. mosquitto
• MQTT 5 connection that, for example, updates device twin, with following config:
  • static specificConfig.clientId
  • specificConfig.cleanSession = false
  • qos 1 or 2 in source

Steps:

• close connection, for example, using HTTP API

  curl -u devops:foobar http://localhost:8080/api/2/connections/<CONN_ID>/command -d connectivity.commands:closeConnection -H 'content-type: text/plain'

• send MQTT 5 message that should lead to device twin updated, with QoS 1 or 2
• open connection, for example, using HTTP API

  curl -u devops:foobar http://localhost:8080/api/2/connections/<CONN_ID>/command -d connectivity.commands:openConnection -H 'content-type: text/plain'

Connection I used to test:

{
    "connectionType": "mqtt-5",
    "connectionStatus": "open",
    "uri": "tcp://mosquitto:1883",
    "specificConfig": {
        "clientId": "ditto",
        "cleanSession": false
    },
    "sources": [
        {
            "addresses": [ "data" ],
            "qos": 2,
            "authorizationContext": ["nginx:ditto"],
            "payloadMapping": ["js-data"]
        },
        {
            "addresses": [ "ditto" ],
            "qos": 2,
            "authorizationContext": ["nginx:ditto"]
        }
    ],
    "targets": [],
    "mappingDefinitions": {
        "js-data": {
            "mappingEngine": "JavaScript",
            "options": {
                "incomingScript": " function mapToDittoProtocolMsg( headers, textPayload, bytePayload, contentType) { return Ditto.buildDittoProtocolMsg( 'test', 'thing-01', 'things', 'twin', 'commands', 'merge', '/features/mqtt/properties/data', { 'content-type': 'application/merge-patch+json' }, { headers, contentType, textPayload, }); }",
                "outgoingScript": ""
            }
        }
    }
}

Messages I sent:

# update device twin using JS payload mapper
mosquitto_pub -V mqttv5 -t data -m $RANDOM -q 2
# update device twin using Ditto payload mapper
mosquitto_pub -V mqttv5 -t ditto -m '{ "topic": "test/thing-01/things/twin/commands/merge", "headers": {"Content-Type": "application/merge-patch+js
on"}, "path": "/features/mqtt/properties/ditto", "value": { "random": '$RANDOM' } }' -D publish content-type 'application/merge-patch+json' -q 2

[dimabarbul]

I've sent 1 message while connection was closed. Here are logs from mosquitto when I send command openConnection:

1696738117: New connection from 172.18.0.8:58714 on port 1883.
1696738118: Client <unknown> closed its connection.
1696738118: New connection from 172.18.0.8:58728 on port 1883.
1696738118: New client connected from 172.18.0.8:58728 as ditto (p5, c0, k60).
1696738118: No will message specified.
1696738118: Sending CONNACK to ditto (1, 0)
1696738118: Sending PUBLISH to ditto (d0, q2, r0, m7, 'data', ... (4 bytes))
1696738118: Received PUBREC from ditto (Mid: 7)
1696738118: Sending PUBREL to ditto (m7)
1696738118: Received PUBCOMP from ditto (Mid: 7, RC:0)
1696738118: Received SUBSCRIBE from ditto
1696738118: 	data (QoS 2)
1696738118: ditto 2 data
1696738118: Sending SUBACK to ditto
1696738118: Received SUBSCRIBE from ditto
1696738118: 	data (QoS 2)
1696738118: ditto 2 data
1696738118: Sending SUBACK to ditto
1696738118: Received SUBSCRIBE from ditto
1696738118: 	ditto (QoS 2)
1696738118: ditto 2 ditto
1696738118: Sending SUBACK to ditto
1696738118: Received SUBSCRIBE from ditto
1696738118: 	ditto (QoS 2)
1696738118: ditto 2 ditto
1696738118: Sending SUBACK to ditto

According to the logs, messages are delivered and acknowledged before Ditto subscribes to topics. My guess is that it is how MQTT 5 protocol works - when you connect to existing session, you get messages, that are awaiting for you, right after the connection is established.

Another strange thing is the first 2 lines - some connection is opened and closed before connection with client ID ditto is established.

The last one strange thing is that Ditto subscribes twice for each topic.

Those strange things are definitely out of scope of this ticket, but they might provide some clue.

[thjaeckle]

Hi @dimabarbul .
I just saw the 2 issues you created.
Does this one here also happen when there is only 1 source in the connection?

Maybe both issues share the problem with more than 1 source?

[dimabarbul]

Unfortunately, this happens when there is only one source as well.

[thjaeckle]

Both issues are likely bugs in Ditto's MQTT implementation.
This was rewritten to use the reactive API of the HiveMQ MQTT Client in Ditto 3.0.0.

We also faced some problems with the HiveMQ MQTT client, e.g.: hivemq/hivemq-mqtt-client#544
So I would also not rule out a client problem.

Unfortunately, the Ditto committer who is most experienced on the MQTT implementation is no longer contributing actively to Ditto and I currently don't find time to deep dive into it as well.

So I will mark the issues as "help wanted" and hope for contributions.

[dimabarbul]

As the changes are quite risky, it would be great to cover existing and new functionality with tests that ensures correct work with real mqtt broker (I can't think of a way to test this using unit tests), so I'd like to have an integration or system test:

• integration test: we run mqtt broker, e.g., mosquitto, and test some actor/actors with real mqtt connection
• system test: we spin up whole ditto installation, mqtt broker etc. and test the scenarios we need.

I'd like the first approach, because integration test would be faster and we'd be able to cover more scenarios. I see that there is test code that launches mongo DB, so it looks okay to have such kind of tests. Also having tests in this repo will confirm that the functionality works as expected when doing later changes by running PR checks, without need to remember to run system tests manually.
Having system test is also okay, as to me, especially if such base setup already exists (I guess, repo ditto-testing is exactly for this).

@thjaeckle , what do you think will be better approach?

[thjaeckle]

@dimabarbul I would normally use the ditto-testing repo which you pointed out.
The only problem with that is, that we do not have a runner capable of running those tests - the infrastructure by the Eclipse Foundation is too limited to run the system-tests.

The colleagues from Bosch still have an internal CI where they can run the tests.

Locally, you should be able to run the tests as well - however the system-tests for the "connectivity" part of Ditto are quite complex (with a lot of inheritance), so it is also not quite easy ..

Therefore an integration test, spinning up an MQTT broker, would likely be the easier approach.
And you're right, we already have tests spinning up MongoDB as well.