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 3b0af69ff2..c7ec7536d0 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
@@ -241,7 +241,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, null, this.mqttPublishActionListener);