Mqttv5PahoMessageDrivenChannelAdapter.addTopic(topic, qos) throws NPE

[mrpiggi]

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

5.5.9 and prior (5.5.5)

Describe the bug

Creating a new Mqttv5PahoMessageDrivenChannelAdapter and using addTopic(topic, qos) before onInit() was called throws a NPE as this.mqttClient.subscribe(topic, qos) is invoked. Same with Mqttv5PahoMessageDrivenChannelAdapter.removeTopic(String...) and this.mqttClient.unsubscribe(topic).

To Reproduce

final Mqttv5PahoMessageDrivenChannelAdapter adapter = new Mqttv5PahoMessageDrivenChannelAdapter(options, id);
adapter.addTopic(topic, 1);

Workaround

final Mqttv5PahoMessageDrivenChannelAdapter adapter = new Mqttv5PahoMessageDrivenChannelAdapter(options, id, topic);
adapter.setQos(1);

Expected behavior

Topic registration or removal without calling IMqttAsyncClient.subscribe(topic, qos) or IMqttAsyncClient.unsubscribe(String...) when not initialized yet.

The comparison between Mqttv5PahoMessageDrivenChannelAdapter and MqttPahoMessageDrivenChannelAdapter is quite obvious.

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

Lines 214 to 242 in 8c57cb7

	@Override
	public void addTopic(String topic, int qos) {
	this.topicLock.lock();
	try {
	this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
	super.addTopic(topic, qos);
	}
	catch (MqttException ex) {
	throw new MessagingException("Failed to subscribe to topic " + topic, ex);
	}
	finally {
	this.topicLock.unlock();
	}
	}
	@Override
	public void removeTopic(String... topic) {
	this.topicLock.lock();
	try {
	this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
	super.removeTopic(topic);
	}
	catch (MqttException ex) {
	throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), ex);
	}
	finally {
	this.topicLock.unlock();
	}
	}

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

Lines 222 to 255 in 8c57cb7

	@Override
	public void addTopic(String topic, int qos) {
	this.topicLock.lock();
	try {
	super.addTopic(topic, qos);
	if (this.client != null && this.client.isConnected()) {
	this.client.subscribe(topic, qos);
	}
	}
	catch (MqttException e) {
	super.removeTopic(topic);
	throw new MessagingException("Failed to subscribe to topic " + topic, e);
	}
	finally {
	this.topicLock.unlock();
	}
	}
	@Override
	public void removeTopic(String... topic) {
	this.topicLock.lock();
	try {
	if (this.client != null && this.client.isConnected()) {
	this.client.unsubscribe(topic);
	}
	super.removeTopic(topic);
	}
	catch (MqttException e) {
	throw new MessagingException("Failed to unsubscribe from topic(s) " + Arrays.toString(topic), e);
	}
	finally {
	this.topicLock.unlock();
	}
	}

[artembilan]

Confirmed.

Feel free to contribute the fix: https://github.com/spring-projects/spring-integration/blob/main/CONTRIBUTING.adoc

Thank you!

[mrpiggi]

I know I'm way too late to the party, but comparing the implementations of MqttPahoMessageDrivenChannelAdapter and Mqttv5PahoMessageDrivenChannelAdapter raises a question: Should topics that throw a MqttException when using mqttClient.subscribe() or mqttClient.unsubscribe() live on in AbstractMqttMessageDrivenChannelAdapter or not? Currently, the implementations are not consistent in this respect.

@artembilan Do you have an opinion on this? If so, I would provide a corresponding PR.

[artembilan]

@mrpiggi ,

I'm not sure in your question, but MqttPahoMessageDrivenChannelAdapter and Mqttv5PahoMessageDrivenChannelAdapter are different with the matter that they use different mqttClient libraries.
We just cannot pull that logic to the super class.

Would you mind to elaborate more what you see could be so generic that it can be pulled to the AbstractMqttMessageDrivenChannelAdapter?

Thanks

[mrpiggi]

@artembilan Sorry for being a bit lazy and unspecific.

Regarding addTopic(String topic, int qos), MqttPahoMessageDrivenChannelAdapter does:

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

Lines 222 to 238 in f013aa8

	@Override
	public void addTopic(String topic, int qos) {
	this.topicLock.lock();
	try {
	super.addTopic(topic, qos);
	if (this.client != null && this.client.isConnected()) {
	this.client.subscribe(topic, qos);
	}
	}
	catch (MqttException e) {
	super.removeTopic(topic);
	throw new MessagingException("Failed to subscribe to topic " + topic, e);
	}
	finally {
	this.topicLock.unlock();
	}
	}

whereas Mqttv5PahoMessageDrivenChannelAdapter does:

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

Lines 221 to 236 in f013aa8

	@Override
	public void addTopic(String topic, int qos) {
	this.topicLock.lock();
	try {
	super.addTopic(topic, qos);
	if (this.mqttClient != null && this.mqttClient.isConnected()) {
	this.mqttClient.subscribe(topic, qos).waitForCompletion(getCompletionTimeout());
	}
	}
	catch (MqttException ex) {
	throw new MessagingException("Failed to subscribe to topic " + topic, ex);
	}
	finally {
	this.topicLock.unlock();
	}
	}

In case of an exception, super.removeTopic() is only invoked by MqttPahoMessageDrivenChannelAdapter but not Mqttv5PahoMessageDrivenChannelAdapter, so for the latter, the topic will remain in Set<Topic> topics within AbstractMqttMessageDrivenChannelAdapter and therefore reflected by getTopic() and getQos().

For removeTopic(String... topic) both adapters call super.removeTopic(topic) after client.unsubscribe(topic), so in case of an exception, these topics will remain in Set<Topic> topics. I did not dig into MqttAsyncClient so I am not sure, if maybe removing the given topics from Set<Topic> topics in any case would be sensible.

Additionally, AbstractMqttMessageDrivenChannelAdapter.addTopic(String... topic) and AbstractMqttMessageDrivenChannelAdapter.addTopic(String[] topic, int[] qos) are not overridden by both MqttPahoMessageDrivenChannelAdapter and Mqttv5PahoMessageDrivenChannelAdapter so calling these methods after initialization won't induce a subscription to the client at all.

[artembilan]

@mrpiggi ,

great observation and investigation!

I'd be glad to see from you a PR for review.

Thanks

[mrpiggi]

My analysis was incomplete. As AbstractMqttMessageDrivenChannelAdapter.addTopic(String... topic) and AbstractMqttMessageDrivenChannelAdapter.addTopic(String[] topic, int[] qos) both call AbstractMqttMessageDrivenChannelAdapter.addTopic(String topic, int qos) and this method is overridden, this aspect is fine.

So, only the question whether to call AbstractMqttMessageDrivenChannelAdapter.removeTopic(String... topic) or not on an Exception is open. Adding topics to both adapters before initialization will be always be reflected by Set<Topic> topics in every case and topics in general are never removed automatically, deleting

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

Line 232 in f013aa8

super.removeTopic(topic);

would probably be the easiest way to retain consistency. Probably not worth a separate PR.

[artembilan]

Why not vise-versa and add super.removeTopic(topic); to the addTopic() of the Mqttv5PahoMessageDrivenChannelAdapter?
I think this would be safer since we know that topic is failed somehow.

I doesn't look, though, that we can track failed topic somehow from the subscribe(String[] topicFilters, int[] qos) call to be able to remove it afterwards...

[mrpiggi]

My intention was to get consistent behavior regardless of whether the adapter has already been initialized or not. If addTopic(String topic, int qos) is used before initialization, the topic is added to Set<Topic> topics in any case and remains in it when doStart() is called, even in case of an exception.

There is also the question of what could actually lead to a MqttException. There is the possibility to query the cause with MqttException.getReasonCode(), but the documentation in org.eclipse.paho for this is rather rudimentary. As far as I can see, a MqttException is only thrown in case of connection problems, which in turn means that when a connection is re-established, it should be possible to retry a subscription for a previously registered topic. Incidentally, an incorrectly formatted topic throws an IllegalArgumentException and thus always remains registered in Set<String> topics.

To work around this, one possibility would be to define a boolean field subscribed within AbstractMqttMessageDrivenChannelAdapter.Topic and set this to true if the call to client.subscribe() does not throw an exception.

[artembilan]

Good. If you have something in mind and can come up with some unit test to verify that behavior, I won't mind to review your contribution.

Thanks