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 330b37aa4d..e35e4df7f2 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2023 the original author or authors.
+ * Copyright 2021-2024 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.mqtt.inbound;
 
 import java.util.Arrays;
 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;
@@ -100,7 +101,7 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 
 	private volatile boolean readyToSubscribeOnStart;
 
-	public Mqttv5PahoMessageDrivenChannelAdapter(String url, String clientId, String... topic) {
+	private final AtomicInteger subscriptionIdentifierCounter = new AtomicInteger(0);	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();
@@ -276,8 +277,10 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 		try {
 			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 }, new MqttProperties())
+								null, null, new IMqttMessageListener[] { this::messageArrived }, subscriptionProperties)
 						.waitForCompletion(getCompletionTimeout());
 			}
 		}
@@ -409,6 +412,10 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 			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())
 					.waitForCompletion(getCompletionTimeout());
 			String message = "Connected and subscribed to " + Arrays.toString(topics);
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 82f696d369..c9bf5d1f32 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
@@ -134,6 +134,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