Incomplete shutdown of Mqttv5PahoMessageDrivenChannelAdapter

[mrpiggi]

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

5.5.6

Describe the bug

Calling Mqttv5PahoMessageDrivenChannelAdapter.doStop() does not disconnect and close the underlying IMqttAsyncClient

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

Lines 187 to 200 in 30bb846

	@Override
	protected void doStop() {
	this.topicLock.lock();
	String[] topics = getTopic();
	try {
	this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
	}
	catch (MqttException ex) {
	logger.error(ex, () -> "Error unsubscribing from " + Arrays.toString(topics));
	}
	finally {
	this.topicLock.unlock();
	}
	}

Expected behavior

I am not sure if the current behavior is intended or a bug. At least, calling MqttPahoMessageDrivenChannelAdapter.doStop() does exactly that

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

Lines 202 to 216 in 30bb846

	try {
	this.client.disconnectForcibly(this.disconnectCompletionTimeout);
	}
	catch (MqttException ex) {
	logger.error(ex, "Exception while disconnecting");
	}
	this.client.setCallback(null);
	try {
	this.client.close();
	}
	catch (MqttException ex) {
	logger.error(ex, "Exception while closing");
	}

If the current behavior is as desired, what do I need to do to disconnect and close the underlying IMqttAsyncClient?

[artembilan]

I think we are missing this.mqttClient.disconnect().waitForCompletion(getDisconnectCompletionTimeout()); alongside with that unsubscribe().
The client.close() has to be done in the destroy() implementation, which we also are missing at the moment.

Thank you! Will fix shortly.