Client keeps disconnecting/reconnecting to broker if it runs as publisher and subscriber at the same time

[hzeng-tfs]

Type of issue

• [X] Bug
• [ ] Enhancement
• [ ] Compliance
• [X] Question
• [ ] Help wanted

Current Behavior

I built a MQTT client application which runs as a publisher and a subscriber at the same time. I follow the ConsoleReferencePublisher and ConsoleReferenceSubsciber basically. I added a few WriterGroup and a ReaderGroup to the the same connection.

// each writer group corresponds to a MQTT topic
foreach (var topic in mDictMqttTopics)
{
    pubSubConnection.WriterGroups.Add(CreateWriterGroup(topic.Value, mqttConfig));
}

pubSubConnection.ReaderGroups.Add(CreateReaderGroup(mDictMqttSubTopics[nameof(ApplyRecipe)], mqttConfig));

//create the PubSub configuration
PubSubConfigurationDataType pubSubConfiguration = new();
pubSubConfiguration.Connections = new PubSubConnectionDataTypeCollection()
    {
        pubSubConnection
    };

It runs correctly with a local Mosquitto server. However, I notice that there are lots of re-connection messages on log of Mosquitto:

20240621T14:22:38: New connection from ::1:1667 on port 1883.
20240621T14:22:38: New client connected from ::1:1667 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:38: No will message specified.
20240621T14:22:38: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:38: New connection from ::1:1668 on port 1883.
20240621T14:22:38: Client ClientId_0827362928 already connected, closing old connection.
20240621T14:22:38: New client connected from ::1:1668 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:38: No will message specified.
20240621T14:22:38: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:38: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:38:  Json_WriterGroup_1 (QoS 0)
20240621T14:22:38: ClientId_0827362928 0 Json_WriterGroup_1
20240621T14:22:38: Sending SUBACK to ClientId_0827362928
20240621T14:22:38: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:38:  Json_WriterGroup_1/$Metadata (QoS 0)
20240621T14:22:38: ClientId_0827362928 0 Json_WriterGroup_1/$Metadata
20240621T14:22:38: Sending SUBACK to ClientId_0827362928
20240621T14:22:43: New connection from ::1:1670 on port 1883.
20240621T14:22:43: Client ClientId_0827362928 already connected, closing old connection.
20240621T14:22:43: New client connected from ::1:1670 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:43: No will message specified.
20240621T14:22:43: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:46: Received PUBLISH from ClientId_0827362928 (d0, q1, r0, m1, 'MySite/MyProduct_1/System', ... (395 bytes))
20240621T14:22:46: Sending PUBACK to ClientId_0827362928 (m1, rc16)
20240621T14:22:48: New connection from ::1:1672 on port 1883.
20240621T14:22:48: Client ClientId_0827362928 already connected, closing old connection.
20240621T14:22:48: New client connected from ::1:1672 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:48: No will message specified.
20240621T14:22:48: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:48: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:48:  Json_WriterGroup_1 (QoS 0)
20240621T14:22:48: ClientId_0827362928 0 Json_WriterGroup_1
20240621T14:22:48: Sending SUBACK to ClientId_0827362928
20240621T14:22:48: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:48:  Json_WriterGroup_1/$Metadata (QoS 0)
20240621T14:22:48: ClientId_0827362928 0 Json_WriterGroup_1/$Metadata
20240621T14:22:48: Sending SUBACK to ClientId_0827362928
20240621T14:22:53: New connection from ::1:1675 on port 1883.
20240621T14:22:53: Client ClientId_0827362928 already connected, closing old connection.
20240621T14:22:53: New client connected from ::1:1675 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:53: No will message specified.
20240621T14:22:53: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:54: Received PUBLISH from ClientId_0827362928 (d0, q1, r0, m1, 'MySite/MyProduct_1/System', ... (395 bytes))
20240621T14:22:54: Sending PUBACK to ClientId_0827362928 (m1, rc16)
20240621T14:22:58: New connection from ::1:1679 on port 1883.
20240621T14:22:58: Client ClientId_0827362928 already connected, closing old connection.
20240621T14:22:58: New client connected from ::1:1679 as ClientId_0827362928 (p5, c1, k15).
20240621T14:22:58: No will message specified.
20240621T14:22:58: Sending CONNACK to ClientId_0827362928 (0, 0)
20240621T14:22:58: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:58:  Json_WriterGroup_1 (QoS 0)
20240621T14:22:58: ClientId_0827362928 0 Json_WriterGroup_1
20240621T14:22:58: Sending SUBACK to ClientId_0827362928
20240621T14:22:58: Received SUBSCRIBE from ClientId_0827362928
20240621T14:22:58:  Json_WriterGroup_1/$Metadata (QoS 0)
20240621T14:22:58: ClientId_0827362928 0 Json_WriterGroup_1/$Metadata
20240621T14:22:58: Sending SUBACK to ClientId_0827362928
20240621T14:23:02: Received DISCONNECT from ClientId_0827362928
20240621T14:23:02: Client ClientId_0827362928 disconnected.

On MQTT client side, the logs indicate that SessionTakenOver is the cause.

06/21/2024 14:22:38.329 MqttPubSubConnection with name 'Publisher/Subscriber MQTT Json' was created.
06/21/2024 14:22:38.343 UaPubSubApplication is starting.
06/21/2024 14:22:38.414 The provided MQTT message handler is null therefore messages will not be processed on client (null)!!!
06/21/2024 14:22:38.416 The provided MQTT message topic filter is null therefore messages will not be processed on client (null)!!!
06/21/2024 14:22:38.473 MQTT client ClientId_0827362928 successfully connected
06/21/2024 14:22:38.488 ClientId_0827362928 Connected to MQTTBroker
06/21/2024 14:22:38.506 ClientId_0827362928 Subscribed to topics: Json_WriterGroup_1,Json_WriterGroup_1/$Metadata
06/21/2024 14:22:38.509 MQTT client ClientId_0827362928 successfully connected
06/21/2024 14:22:38.511 Connection 'Publisher/Subscriber MQTT Json' started 5 publishers and 1 subscribers.
06/21/2024 14:22:38.513 Connection 'Publisher/Subscriber MQTT Json' was started.

06/21/2024 14:22:43.497 Disconnect Handler called on client ClientId_0827362928, reason: SessionTakenOver was connected: True
06/21/2024 14:22:43.505 MQTT client ClientId_0827362928 successfully connected

06/21/2024 14:22:48.505 Disconnect Handler called on client ClientId_0827362928, reason: SessionTakenOver was connected: True
06/21/2024 14:22:48.520 ClientId_0827362928 Connected to MQTTBroker
06/21/2024 14:22:48.527 ClientId_0827362928 Subscribed to topics: Json_WriterGroup_1,Json_WriterGroup_1/$Metadata
06/21/2024 14:22:48.529 MQTT client ClientId_0827362928 successfully connected

06/21/2024 14:22:53.513 Disconnect Handler called on client ClientId_0827362928, reason: SessionTakenOver was connected: True
06/21/2024 14:22:53.522 MQTT client ClientId_0827362928 successfully connected

06/21/2024 14:22:58.522 Disconnect Handler called on client ClientId_0827362928, reason: SessionTakenOver was connected: True
06/21/2024 14:22:58.530 ClientId_0827362928 Connected to MQTTBroker
06/21/2024 14:22:58.535 ClientId_0827362928 Subscribed to topics: Json_WriterGroup_1,Json_WriterGroup_1/$Metadata
06/21/2024 14:22:58.537 MQTT client ClientId_0827362928 successfully connected

Expected Behavior

the reconnection should not happen.

I debugged it a little and I think the cause of the issue is on the InternalStart() method of Opc.Ua.PubSub\Transport\MqttPubSubConnection.cs

Two MqttClient instances are created with same ClientId (created on GetMqttClientOptions()), one for publisher and one for subscriber. When two MQTT clients with same ClientId connect to a broker, the broker just connects the new one and disconnect existing one.

MqttClient publisherClient = null;
MqttClient subscriberClient = null;
MqttClientOptions mqttOptions = GetMqttClientOptions();

int nrOfPublishers = Publishers.Count;
int nrOfSubscribers = GetAllDataSetReaders().Count;

//publisher initialization
if (nrOfPublishers > 0)
{
    publisherClient = (MqttClient)await MqttClientCreator.GetMqttClientAsync(
        m_reconnectIntervalSeconds,
        mqttOptions,
        null).ConfigureAwait(false);
}

//subscriber initialization
if (nrOfSubscribers > 0)
{
    // collect all topics from all ReaderGroups
    var topics = new StringCollection();
    foreach (ReaderGroupDataType readerGroup in PubSubConnectionConfiguration.ReaderGroups)
    {
        if (!readerGroup.Enabled)
        {
            continue;
        }

        foreach (DataSetReaderDataType dataSetReader in readerGroup.DataSetReaders)
        {
            if (!dataSetReader.Enabled)
            {
                continue;
            }

            if (ExtensionObject.ToEncodeable(dataSetReader.TransportSettings) is BrokerDataSetReaderTransportDataType brokerTransportSettings && !topics.Contains(brokerTransportSettings.QueueName))
            {
                topics.Add(brokerTransportSettings.QueueName);

                if (brokerTransportSettings.MetaDataQueueName != null)
                {
                    topics.Add(brokerTransportSettings.MetaDataQueueName);
                }
            }
        }
    }

    subscriberClient = (MqttClient)await MqttClientCreator.GetMqttClientAsync(
        m_reconnectIntervalSeconds,
        mqttOptions,
        ProcessMqttMessage,
        topics).ConfigureAwait(false);
}

The 5s pattern on log file is caused by private int m_reconnectIntervalSeconds = 5; I think.

So the potential fix:

• one MqttClient instance for publisher and subscriber. (mqttMultimeter use one instance https://github.com/chkr1011/mqttMultimeter/blob/main/Source/Services/Mqtt/MqttClientService.cs )
• two MqttClient instances, one for each, but has different ClientId

Steps To Reproduce

No response

Environment

- OS: Windows 10
- Environment:
- Runtime:
- Nuget Version: 1.5.374.70-beta PubSub
- Component: PubSub
- Server:
- Client:

Anything else?

As I am still new on MQTT area, please let me know if there is an option to solve this behaviour or I did the coding wrong. Thanks.

[hzeng-tfs]

I temporarily solved the issue by using two connections, one for publisher and one for subscriber to make sure they have different ClientIds. This confirmed my observation.

Since a connection is allowed to have a publisher and a subscriber, I think it is a bug.

by the way, it could be nice to add an option to set ClientId explicitly rather than using a random number only.

[mrsuciu]

@hzeng-tfs Yes this is a bug, even the ClientId documentation states "Hint: This identifier needs to be unique over all used clients / devices on the broker to avoid connection issues."