diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java
index 60ce021d45..6189e6c644 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/Mqttv5PahoMessageHandler.java
@@ -266,7 +266,15 @@ public class Mqttv5PahoMessageHandler extends AbstractMqttMessageHandler<IMqttAs
 		long completionTimeout = getCompletionTimeout();
 		try {
 			if (!this.mqttClient.isConnected()) {
-				this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout);
+				this.lock.lock();
+				try {
+					if (!this.mqttClient.isConnected()) {
+						this.mqttClient.connect(this.connectionOptions).waitForCompletion(completionTimeout);
+					}
+				}
+				finally {
+					this.lock.unlock();
+				}
 			}
 			IMqttToken token = this.mqttClient.publish(topic, (MqttMessage) mqttMessage);
 			ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher();