From fb6e6c63df3bd092bd5b18b3412f440d2dcffcf3 Mon Sep 17 00:00:00 2001
From: Artem Bilan <artem.bilan@broadcom.com>
Date: Fri, 25 Oct 2024 16:52:27 -0400
Subject: [PATCH] GH-9597: Fix `Mqttv5PahoMessageHandler.publish()` for
 concurrency

Fixes: #9597
Issue link: https://github.com/spring-projects/spring-integration/issues/9597

When input channel for the endpoint with a `Mqttv5PahoMessageHandler` is an `ExecutorChannel` and a lot of concurrent messages are sent,
the `Connect already in progress (32110)` exception can be thrown

* Wrap `this.mqttClient.connect()` in the `Mqttv5PahoMessageHandler.publish()` into a `Lock`

(cherry picked from commit 334620f36ea7c2c576d76edbe7669f3bf8044e60)
---
 .../mqtt/outbound/Mqttv5PahoMessageHandler.java        | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

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();