GH-4002: Deprecate ConsumerStopAction in MQTT (#4006)

Fixes #4002

Turns out the `ConsumerStopAction` was introduced in a point version to mitigate an unsubscription bug
and let to preserve a previous behaviour.

* Deprecate `ConsumerStopAction` in favor of just `cleanSession` flag in the `MqttConnectOptions`

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

Author: artembilan

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/ConsumerStopAction.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2023 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.
@@ -17,12 +17,17 @@
 package org.springframework.integration.mqtt.core;
 
 /**
- * Action to take regarding subscrptions when consumer stops.
+ * Action to take regarding subscriptions when consumer stops.
  *
  * @author Gary Russell
+ *
  * @since 4.2.3
  *
+ * @deprecated since 5.5.17
+ * in favor of standard {@link org.eclipse.paho.client.mqttv3.MqttConnectOptions#setCleanSession(boolean)}.
+ * Will be removed in 6.1.0.
  */
+@Deprecated
 public enum ConsumerStopAction {
 
 	/**

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/DefaultMqttPahoClientFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -41,6 +41,7 @@ public class DefaultMqttPahoClientFactory implements MqttPahoClientFactory {
 
 	private MqttClientPersistence persistence;
 
+	@SuppressWarnings("deprecation")
 	private ConsumerStopAction consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN;
 
 	/**
@@ -55,7 +56,10 @@ public void setPersistence(MqttClientPersistence persistence) {
 	 * Get the consumer stop action.
 	 * @return the consumer stop action.
 	 * @since 4.2.3
+	 * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}.
+	 * Will be removed in 6.1.0.
 	 */
+	@Deprecated
 	@Override
 	public ConsumerStopAction getConsumerStopAction() {
 		return this.consumerStopAction;
@@ -66,7 +70,10 @@ public ConsumerStopAction getConsumerStopAction() {
 	 * Default: {@link ConsumerStopAction#UNSUBSCRIBE_CLEAN}.
 	 * @param consumerStopAction the consumer stop action.
 	 * @since 4.2.3.
+	 * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}.
+	 * Will be removed in 6.1.0.
 	 */
+	@Deprecated
 	public void setConsumerStopAction(ConsumerStopAction consumerStopAction) {
 		this.consumerStopAction = consumerStopAction;
 	}

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/core/MqttPahoClientFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -60,7 +60,10 @@ public interface MqttPahoClientFactory {
 	 * Get the consumer stop action.
 	 * @return the consumer stop action.
 	 * @since 4.3
+	 * @deprecated since 5.5.17 in favor of standard {@link MqttConnectOptions#setCleanSession(boolean)}.
+	 * Will be removed in 6.1.0.
 	 */
+	@Deprecated
 	ConsumerStopAction getConsumerStopAction();
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -31,7 +31,6 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.acks.SimpleAcknowledgment;
-import org.springframework.integration.mqtt.core.ConsumerStopAction;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
 import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
 import org.springframework.integration.mqtt.core.MqttPahoComponent;
@@ -83,7 +82,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 
 	private volatile boolean cleanSession;
 
-	private volatile ConsumerStopAction consumerStopAction;
+	@SuppressWarnings("deprecation")
+	private volatile org.springframework.integration.mqtt.core.ConsumerStopAction consumerStopAction;
 
 	/**
 	 * Use this constructor for a single url (although it may be overridden if the server
@@ -184,13 +184,16 @@ protected void doStart() {
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	@Override
 	protected synchronized void doStop() {
 		cancelReconnect();
 		if (this.client != null) {
 			try {
-				if (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
-						|| (this.consumerStopAction.equals(ConsumerStopAction.UNSUBSCRIBE_CLEAN)
+				if (this.consumerStopAction
+						.equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_ALWAYS)
+						|| (this.consumerStopAction
+						.equals(org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN)
 						&& this.cleanSession)) {
 
 					this.client.unsubscribe(getTopic());
@@ -254,12 +257,13 @@ public void removeTopic(String... topic) {
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	private synchronized void connectAndSubscribe() throws MqttException { // NOSONAR
 		MqttConnectOptions connectionOptions = this.clientFactory.getConnectionOptions();
 		this.cleanSession = connectionOptions.isCleanSession();
 		this.consumerStopAction = this.clientFactory.getConsumerStopAction();
 		if (this.consumerStopAction == null) {
-			this.consumerStopAction = ConsumerStopAction.UNSUBSCRIBE_CLEAN;
+			this.consumerStopAction = org.springframework.integration.mqtt.core.ConsumerStopAction.UNSUBSCRIBE_CLEAN;
 		}
 		Assert.state(getUrl() != null || connectionOptions.getServerURIs() != null,
 				"If no 'url' provided, connectionOptions.getServerURIs() must not be null");

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

@@ -18,7 +18,6 @@
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
-import java.util.Date;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -62,7 +61,6 @@
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.MessageProcessor;
-import org.springframework.integration.mqtt.core.ConsumerStopAction;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
 import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
 import org.springframework.integration.mqtt.event.MqttIntegrationEvent;
@@ -124,7 +122,7 @@ public class MqttAdapterTests {
 	public void testCloseOnBadConnectIn() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
 		willThrow(new MqttException(0)).given(client).connect(any());
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null);
 		adapter.start();
 		verify(client).close();
 		adapter.stop();
@@ -370,7 +368,7 @@ public Message<?> toMessage(Object payload, MessageHeaders headers) {
 	@Test
 	public void testStopActionDefault() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, null);
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null);
 
 		adapter.start();
 		adapter.stop();
@@ -380,35 +378,7 @@ public void testStopActionDefault() throws Exception {
 	@Test
 	public void testStopActionDefaultNotClean() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false, null);
-
-		adapter.start();
-		adapter.stop();
-		verifyNotUnsubscribe(client);
-	}
-
-	@Test
-	public void testStopActionAlways() throws Exception {
-		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false,
-				ConsumerStopAction.UNSUBSCRIBE_ALWAYS);
-
-		adapter.start();
-		adapter.stop();
-		verifyUnsubscribe(client);
-
-		adapter.connectionLost(new RuntimeException("Intentional"));
-
-		TaskScheduler taskScheduler = TestUtils.getPropertyValue(adapter, "taskScheduler", TaskScheduler.class);
-
-		verify(taskScheduler, never())
-				.schedule(any(Runnable.class), any(Date.class));
-	}
-
-	@Test
-	public void testStopActionNever() throws Exception {
-		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, false);
 
 		adapter.start();
 		adapter.stop();
@@ -439,7 +409,7 @@ public void testCustomExpressions() {
 	@Test
 	public void testReconnect() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null);
 		adapter.setRecoveryInterval(10);
 		LogAccessor logger = spy(TestUtils.getPropertyValue(adapter, "logger", LogAccessor.class));
 		new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
@@ -576,7 +546,7 @@ public void testDifferentQos() throws Exception {
 	@Test
 	public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
-		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null);
 		adapter.setRecoveryInterval(10);
 
 		MqttException mqttException = new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED);
@@ -615,8 +585,8 @@ public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
 		taskScheduler.destroy();
 	}
 
-	private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession,
-			ConsumerStopAction action) {
+	private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession)
+			throws MqttException {
 
 		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
 
@@ -631,9 +601,6 @@ public IMqttClient getClientInstance(String uri, String clientId) throws MqttExc
 		if (cleanSession != null) {
 			connectOptions.setCleanSession(cleanSession);
 		}
-		if (action != null) {
-			factory.setConsumerStopAction(action);
-		}
 		factory.setConnectionOptions(connectOptions);
 		given(client.isConnected()).willReturn(true);
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("client", factory, "foo");