Shared subscription prefix "$SharedSubscription" not supported

[michelblk]

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

Since version 6.5.6

Describe the bug

We use the mqtt topic prefix "$SharedSubscription" instead of the default prefix "$share" to use to mqtt shared subscription feature.

Since version 6.5.6 of spring-integration-mqtt, all MQTTv5 messages are ignored when the "$SharedSubscription" prefix is used. The topics of incoming messages do not match any subscribed topics, so no messages are processed.

The relevant code appears to be:

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

Line 444 in 79f6938

|| subscribedTopic.startsWith("$share/") && subscribedTopic.endsWith(receivedTopic)) {

The topic prefix for shared subscriptions is currently not configurable. This leads to incompatibility with brokers like IBM MessageSight, which require "$SharedSubscription".

To Reproduce

1. Use an MQTT broker that requires the "$SharedSubscription" prefix for shared subscriptions (e.g. IBM MessageSight).
2. Configure spring-integration-mqtt to subscribe to "$SharedSubscription/prod/myTopic".
3. Send a message to "myTopic".
4. The callback function is not triggered because the topic is not recognized.

Expected behavior

The callback function should be triggered when a message arrives on the topic "myTopic"—regardless of whether the prefix is "$share" or "$SharedSubscription". The topic prefix should be configurable.

Sample

[michelblk]

My bad, it seems like that this behaviour was already fixed with version 6.5.7. Thank you!

[mrpiggi]

Are you sure? There was a fix regarding shared subscriptions but the prefix "$share/" is still hard coded with no option to configure:

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

Lines 537 to 549 in dd781b3

	private void messageArrivedIfMatched(String topic, MqttMessage mqttMessage) {
	for (String subscribedTopic : getTopic()) {
	if (subscribedTopic.startsWith("$share/")) {
	subscribedTopic = subscribedTopic.split("/", 3)[2];
	}
	if (MqttTopicValidator.isMatched(subscribedTopic, topic)) {
	messageArrived(topic, mqttMessage);
	return;
	}
	}
	logger.trace(() ->
	"Arrived message on topic '" + topic + "' this channel adapter is not subscribed to. Ignoring...");
	}

[artembilan]

Sorry, I didn’t dig too deep for this issue after realizing it is closed as duplication .

So, if we agree that shared subscription prefix could be anything, then give me please a clue what exactly pattern in MQTT spec has to be used to indicate that topic is in shared subscription?

I mean it is a bit frustrating that these details have to be mitigated here in Spring Integration instead of that Paho library which can be used anywhere . But the reality is like we have so far. Therefore I want help from the community to bring as robust solution as possible .

Thanks and reopening issue respectively