Subscription Identifier always included when adding an mqtt topic on MQTT 5

[MarcelFadtke]

In what version(s) of Spring Integration are you seeing this issue?

6.3.0.RELEASE

(likely since / caused by #8879)

Describe the bug

When adding a topic to the subscriptions, the following section causes the inclusion of a subscription identifier even if the server sent that the subscription identifiers are not available (during the connection by the Connect Ack response) or if the paho mqtt client connection options are configured to not use subscription identifiers.

Code:

spring-integration/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

Lines 346 to 350 in 91de12c

	MqttProperties subscriptionProperties = new MqttProperties();
	subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());
	this.mqttClient.subscribe(new MqttSubscription[] {subscription},
	null, null, new IMqttMessageListener[] {this::messageArrived}, subscriptionProperties)
	.waitForCompletion(getCompletionTimeout());

To Reproduce

Setup an mqtt 5 client and connect to a broker and adding a topic programatically, while tracking the mqtt message properties with wireshark.

Expected behavior

The method includes mqtt connection settings from the server's Connect Ack message and the configured client MqttConnectionOption, so that the subscription identifier usage can be disabled / turned off.

Sample

A link to a GitHub repository with a minimal, reproducible sample.

Reports that include a sample will take priority over reports that do not.
At times, we may require a sample, so it is good to try and include a sample up front.

[artembilan]

I'm not fully understand the report.
Do you mean that this identifier has to be optional?
What condition we can use for that?
We have added this identifier as a result of this issue: #8879
Thanks

[MarcelFadtke]

Hi @artembilan and thank you for the quick response,

what I mean is that the subscription identifier must not be used in the subscription message from the client if the broker does not support subscription identifiers.

Just to make sure, this issues addresses only the usage of mqtt 5 connections.

According to the specification of mqtt5, section '3.2.2.3.12 Subscription Identifiers Available' (https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901092) the broker can support the usage of subscription identifiers but it's not a must.
If the mqtt client subscribes to a topic with a subscription identifier in the message parameters () the server will respond with an error 161 which you can find in the spec here 3.9.3 SUBACK Payload (https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc3901178)

In addition there is a configuration option in the MqttConnectionOptions of the mqtt5 paho client, that allows to configure the usage of subscription identifiers (https://github.com/eclipse/paho.mqtt.java/blob/master/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttConnectionOptions.java#L64 // https://github.com/eclipse/paho.mqtt.java/blob/master/org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/MqttConnectionOptions.java#L894).
I'm not sure about the connection between that configuration option in the paho mqtt library and spring integration, but I guess, that this should be considered as well.
So if I configure the MqttConnectionOptions of the paho client that the usages of subscription identifiers is disabled, spring also shouldn't set them.

In my scenario I connected to an mqtt 5 broker which does not support the usage of subscription identifiers. I also configured the MqttConnectionOptions to not use subscription identifiers. Yet the subscription message contains the parameter of the subscription identifier, when I used the method for adding a topic.

Best regards,
Marcel

[artembilan]

OK. I see the logic:

		int subId = subscriptionProperties.getSubscriptionIdentifiers().get(0);

		// Automatic Subscription Identifier Assignment is enabled
		if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {

			// Application is overriding the subscription Identifier
			if (subId != 0) {
				// Check that we are not already using this ID, else throw Illegal Argument
				// Exception
				if (this.comms.doesSubscriptionIdentifierExist(subId)) {
					throw new IllegalArgumentException(
							String.format("The Subscription Identifier %s already exists.", subId));
				}

			} else {
				// Automatically assign new ID and link to callback.
				subId = this.mqttSession.getNextSubscriptionIdentifier();
			}
		}

I wonder then why have we introduced that subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());.

Looks like it is a default behavior anyway if this.mqttConnection.isSubscriptionIdentifiersAvailable().

Testing...

[artembilan]

OK. So, we just used wrong API to subscribe.
This one makes our tests happy:

			MqttProperties subscriptionProperties = new MqttProperties();
			// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
			subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
			this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, subscriptionProperties)

And if you choose useSubscriptionIdentifiers = false in those MqttConnectionOptions, the no identifier are going to happen.
If I understand that Paho logic well...

Will provide the fix shortly.

[aditya-pathak-dc]

@artembilan @wilkinsona @spring-builds @MarcelFadtke I am using 6.3.0 version of spring-mqtt-integration, but still facing the same issue. Unable to receive message on shared topics.
Here are the details: https://stackoverflow.com/questions/78605910/spring-mqtt-integration-shared-subscritopn-is-not-working