diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
index cdb59906e2..acc3e04e04 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
@@ -21,6 +21,7 @@ import java.util.Arrays;
 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;
@@ -107,6 +108,8 @@ 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.
@@ -336,8 +339,10 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 				this.subscriptions.add(subscription);
 			}
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
+				MqttProperties subscriptionProperties = new MqttProperties();
+				subscriptionProperties.setSubscriptionIdentifier(this.subscriptionIdentifierCounter.incrementAndGet());
 				this.mqttClient.subscribe(new MqttSubscription[] { subscription },
-								null, null, new IMqttMessageListener[] { this::messageArrived }, new MqttProperties())
+								null, null, new IMqttMessageListener[] { this::messageArrived }, subscriptionProperties)
 						.waitForCompletion(getCompletionTimeout());
 			}
 		}
@@ -467,7 +472,11 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 			IMqttMessageListener[] listeners = IntStream.range(0, mqttSubscriptions.length)
 					.mapToObj(t -> listener)
 					.toArray(IMqttMessageListener[]::new);
-			this.mqttClient.subscribe(mqttSubscriptions, null, null, listeners, new MqttProperties())
+			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)
 					.waitForCompletion(getCompletionTimeout());
 			String message = "Connected and subscribed to " + Arrays.toString(mqttSubscriptions);
 			logger.debug(message);
diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java
index 80e596d0cf..588d62d8e5 100644
--- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java
+++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java
@@ -137,6 +137,22 @@ public class Mqttv5BackToBackTests implements MosquittoContainerTest {
 		assertThat(receive.getPayload()).isEqualTo(testPayload);
 	}
 
+	@Test
+	public void testSharedTopicMqttv5Interaction() {
+		this.mqttv5MessageDrivenChannelAdapter.addTopic("$share/group/testTopic");
+
+		String testPayload = "shared topic payload";
+		this.mqttOutFlowInput.send(
+				MessageBuilder.withPayload(testPayload)
+						.setHeader(MqttHeaders.TOPIC, "testTopic")
+						.build());
+
+		Message<?> receive = this.fromMqttChannel.receive(10_000);
+
+		assertThat(receive).isNotNull();
+		assertThat(receive.getPayload()).isEqualTo(testPayload);
+	}
+
 
 	@Configuration
 	@EnableIntegration