GH-3955: Mqtt adapter unsubscribe when cleanStart

Fixes #3955

`Mqttv5PahoMessageDrivenChannelAdapter` unsubscribes from all topics, even if `cleanStart/cleanSession` is set to `false`, thus not receiving offline messages after restart.

* unsubscribe `Mqttv5PahoMessageDrivenChannelAdapter` only when `cleanStart`
* add tests

**Cherry-pick to `5.5.x`**

Author: mths1

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

@@ -72,6 +72,7 @@
  * @author Mikhail Polivakha
  * @author Lucas Bowler
  * @author Artem Vozhdayenko
+ * @author Matthias Thoma
  *
  * @since 5.5.5
  *
@@ -217,7 +218,9 @@ protected void doStop() {
 		String[] topics = getTopic();
 		try {
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
-				this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+				if (this.connectionOptions.isCleanStart()) {
+					this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+				}
 
 				if (getClientManager() == null) {
 					this.mqttClient.disconnect().waitForCompletion(getCompletionTimeout());

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5AdapterTests.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2002-2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.mqtt;
+
+
+import org.eclipse.paho.mqttv5.client.IMqttAsyncClient;
+import org.eclipse.paho.mqttv5.client.IMqttMessageListener;
+import org.eclipse.paho.mqttv5.client.IMqttToken;
+import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
+import org.eclipse.paho.mqttv5.common.MqttException;
+import org.eclipse.paho.mqttv5.common.MqttSubscription;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.integration.channel.NullChannel;
+import org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * @author Gary Russell
+ * @author Artem Bilan
+ * @author Artem Vozhdayenko
+ * @author Matthias Thoma
+ *
+ * @since 5.5.16
+ *
+ */
+public class Mqttv5AdapterTests {
+
+	@Test
+	public void testStop() throws Exception {
+		final IMqttAsyncClient client = mock(IMqttAsyncClient.class);
+		Mqttv5PahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, true);
+
+		adapter.start();
+		adapter.connectComplete(false, null);
+		adapter.stop();
+
+		verify(client).connect(any(MqttConnectionOptions.class));
+		verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
+		verify(client).unsubscribe(any(String[].class));
+	}
+
+	@Test
+	public void testStopNotClean() throws Exception {
+		final IMqttAsyncClient client = mock(IMqttAsyncClient.class);
+		Mqttv5PahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false);
+
+		adapter.start();
+		adapter.connectComplete(false, null);
+		adapter.stop();
+
+		verify(client).connect(any(MqttConnectionOptions.class));
+		verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
+		verify(client, never()).unsubscribe(any(String[].class));
+	}
+
+	private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttAsyncClient client, boolean cleanStart) throws MqttException {
+
+		MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
+		connectionOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
+		connectionOptions.setCleanStart(cleanStart);
+
+		given(client.isConnected()).willReturn(true);
+		IMqttToken token = mock(IMqttToken.class);
+		given(client.disconnect()).willReturn(token);
+		given(client.connect(any(MqttConnectionOptions.class))).willReturn(token);
+		given(client.subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any())).willReturn(token);
+		given(client.unsubscribe(any(String[].class))).willReturn(token);
+		Mqttv5PahoMessageDrivenChannelAdapter adapter = new Mqttv5PahoMessageDrivenChannelAdapter(connectionOptions, "client", "foo");
+		ReflectionTestUtils.setField(adapter, "mqttClient", client);
+		adapter.setBeanFactory(mock(BeanFactory.class));
+		adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));
+		adapter.setOutputChannel(new NullChannel());
+		adapter.afterPropertiesSet();
+		return adapter;
+	}
+
+}