GH-9214: Fix subscription identifier logic in Mqttv5PahoMessageDrivenChannelAdapter

Fixes: #9214

The `MqttAsyncClient` can set `subscriptionIdentifier` from session if it is enabled
and available from connection

* Remove manual `subscriptionIdentifierCounter` from the `Mqttv5PahoMessageDrivenChannelAdapter`.
Instead, use `subscriptionProperties.setSubscriptionIdentifiers(List.of(0));`
to make the `this.mqttSession.getNextSubscriptionIdentifier();` to work
when `if (connOpts.useSubscriptionIdentifiers() && this.mqttConnection.isSubscriptionIdentifiersAvailable()) {` is `true`

(cherry picked from commit 3dcbdef)

Author: artembilan

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

@@ -21,13 +21,11 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.IntStream;
 
 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.MqttAsyncClient;
 import org.eclipse.paho.mqttv5.client.MqttCallback;
@@ -109,8 +107,6 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 
 	private volatile boolean readyToSubscribeOnStart;
 
-	private final AtomicInteger subscriptionIdentifierCounter = new AtomicInteger(0);
-
 	/**
 	 * Create an instance based on the MQTT url, client id and subscriptions.
 	 * @param url the MQTT url to connect.
@@ -344,9 +340,10 @@ public void addTopic(String topic, int qos) {
 			}
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
 				MqttProperties subscriptionProperties = new MqttProperties();
-				subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());
+				// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
+				subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
 				this.mqttClient.subscribe(new MqttSubscription[] {subscription},
-								null, null, new IMqttMessageListener[] {this::messageArrived}, subscriptionProperties)
+								null, null, this::messageArrived, subscriptionProperties)
 						.waitForCompletion(getCompletionTimeout());
 			}
 		}
@@ -472,15 +469,10 @@ private void subscribe() {
 		ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();
 		this.topicLock.lock();
 		try {
-			IMqttMessageListener listener = this::messageArrived;
-			IMqttMessageListener[] listeners = IntStream.range(0, mqttSubscriptions.length)
-					.mapToObj(t -> listener)
-					.toArray(IMqttMessageListener[]::new);
 			MqttProperties subscriptionProperties = new MqttProperties();
-			subscriptionProperties.setSubscriptionIdentifiers(IntStream.range(0, mqttSubscriptions.length)
-					.mapToObj(i -> this.subscriptionIdentifierCounter.incrementAndGet())
-					.toList());
-			this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, subscriptionProperties)
+			// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
+			subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
+			this.mqttClient.subscribe(mqttSubscriptions, null, null, this::messageArrived, subscriptionProperties)
 					.waitForCompletion(getCompletionTimeout());
 			String message = "Connected and subscribed to " + Arrays.toString(mqttSubscriptions);
 			logger.debug(message);

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

@@ -57,7 +57,7 @@ public void testStop() throws Exception {
 		adapter.stop();
 
 		verify(client).connect(any(MqttConnectionOptions.class));
-		verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
+		verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener.class), any());
 		verify(client).unsubscribe(any(String[].class));
 	}
 
@@ -71,11 +71,12 @@ public void testStopNotClean() throws Exception {
 		adapter.stop();
 
 		verify(client).connect(any(MqttConnectionOptions.class));
-		verify(client).subscribe(any(MqttSubscription[].class), any(), any(), any(IMqttMessageListener[].class), any());
+		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 {
+	private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttAsyncClient client,
+			boolean cleanStart) throws MqttException {
 
 		MqttConnectionOptions connectionOptions = new MqttConnectionOptions();
 		connectionOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
@@ -85,9 +86,11 @@ private static Mqttv5PahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttA
 		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.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");
+		Mqttv5PahoMessageDrivenChannelAdapter adapter =
+				new Mqttv5PahoMessageDrivenChannelAdapter(connectionOptions, "client", "foo");
 		ReflectionTestUtils.setField(adapter, "mqttClient", client);
 		adapter.setBeanFactory(mock(BeanFactory.class));
 		adapter.setApplicationEventPublisher(mock(ApplicationEventPublisher.class));