GH-9276: Mitigate ConcurrentModificationException in the Mqttv5PahoMessageDrivenChannelAdapter

Fixes: #9276

The current Eclipse Paho client has wrong removal from map logic which leads to the
`ConcurrentModificationException` when we unsubscribe from topic.

* Catch a `ConcurrentModificationException` `this.mqttClient.unsubscribe()` and just log it as an `error`

**Auto-cherry-pick to `6.3.x` & `6.2.x`**

Author: artembilan

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.ConcurrentModificationException;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.locks.Lock;
@@ -289,7 +290,7 @@ protected void doStop() {
 		try {
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
 				if (this.connectionOptions.isCleanStart()) {
-					this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+					unsubscribe(topics);
 					// Have to re-subscribe on next start if connection is not lost.
 					this.readyToSubscribeOnStart = true;
 
@@ -310,12 +311,22 @@ protected void doStop() {
 		}
 	}
 
+	private void unsubscribe(String... topics) throws MqttException {
+		try {
+			// Catch ConcurrentModificationException: https://github.com/eclipse/paho.mqtt.java/issues/986
+			this.mqttClient.unsubscribe(topics).waitForCompletion(getCompletionTimeout());
+		}
+		catch (ConcurrentModificationException ex) {
+			logger.error(ex, () -> "Error unsubscribing from " + Arrays.toString(topics));
+		}
+	}
+
 	@Override
 	public void destroy() {
 		super.destroy();
 		try {
 			if (getClientManager() == null && this.mqttClient != null) {
-				this.mqttClient.close(true);
+				this.mqttClient.close();
 			}
 		}
 		catch (MqttException ex) {
@@ -360,7 +371,7 @@ public void removeTopic(String... topic) {
 		this.topicLock.lock();
 		try {
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
-				this.mqttClient.unsubscribe(topic).waitForCompletion(getCompletionTimeout());
+				unsubscribe(topic);
 			}
 			super.removeTopic(topic);
 			if (!CollectionUtils.isEmpty(this.subscriptions)) {