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)

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

Author: artembilan

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

@@ -17,14 +17,13 @@
 package org.springframework.integration.mqtt.inbound;
 
 import java.util.Arrays;
+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;
@@ -85,7 +84,7 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 		extends AbstractMqttMessageDrivenChannelAdapter<IMqttAsyncClient, MqttConnectionOptions>
 		implements MqttCallback, MqttComponent<MqttConnectionOptions> {
 
-	private final Lock lock =  new ReentrantLock();
+	private final Lock lock = new ReentrantLock();
 
 	private final MqttConnectionOptions connectionOptions;
 
@@ -102,7 +101,7 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 
 	private volatile boolean readyToSubscribeOnStart;
 
-	private final AtomicInteger subscriptionIdentifierCounter = new AtomicInteger(0);	public Mqttv5PahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
+	public Mqttv5PahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
 		super(url, clientId, topic);
 		Assert.hasText(url, "'url' cannot be null or empty");
 		this.connectionOptions = new MqttConnectionOptions();
@@ -282,9 +281,10 @@ public void addTopic(String topic, int qos) {
 			super.addTopic(topic, qos);
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
 				MqttProperties subscriptionProperties = new MqttProperties();
-				subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());
-				this.mqttClient.subscribe(new MqttSubscription[] { new MqttSubscription(topic, qos) },
-								null, null, new IMqttMessageListener[] { this::messageArrived }, subscriptionProperties)
+				// Make use of mqttSession.getNextSubscriptionIdentifier() if available in connection
+				subscriptionProperties.setSubscriptionIdentifiers(List.of(0));
+				this.mqttClient.subscribe(new MqttSubscription[] {new MqttSubscription(topic, qos)},
+								null, null, this::messageArrived, subscriptionProperties)
 						.waitForCompletion(getCompletionTimeout());
 			}
 		}
@@ -409,18 +409,13 @@ private void subscribe() {
 			}
 
 			int[] requestedQos = getQos();
-			MqttSubscription[] subscriptions = IntStream.range(0, topics.length)
+			MqttSubscription[] mqttSubscriptions = IntStream.range(0, topics.length)
 					.mapToObj(i -> new MqttSubscription(topics[i], requestedQos[i]))
 					.toArray(MqttSubscription[]::new);
-			IMqttMessageListener listener = this::messageArrived;
-			IMqttMessageListener[] listeners = IntStream.range(0, topics.length)
-					.mapToObj(t -> listener)
-					.toArray(IMqttMessageListener[]::new);
 			MqttProperties subscriptionProperties = new MqttProperties();
-			subscriptionProperties.setSubscriptionIdentifiers(IntStream.range(0, topics.length)
-					.mapToObj(i -> this.subscriptionIdentifierCounter.incrementAndGet())
-					.toList());
-			this.mqttClient.subscribe(subscriptions, null, null, listeners, new MqttProperties())
+			// 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(topics);
 			logger.debug(message);
@@ -451,7 +446,6 @@ private static String obtainServerUrlFromOptions(MqttConnectionOptions connectio
 		return serverURIs[0];
 	}
 
-
 	/**
 	 * Used to complete message arrival when {@link #isManualAcks()} is true.
 	 */

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

@@ -58,7 +58,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));
 	}
 
@@ -72,11 +72,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"});
@@ -86,9 +87,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));