Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9276: Mitigate ConcurrentModificationException in the Mqttv5PahoMessageDrivenChannelAdapter

Browse Source 
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`

(cherry picked from commit 12fc353db1)
This commit is contained in:
Artem Bilan
2024-06-26 14:23:53 -04:00
committed by Spring Builds
parent 04fac20f90
commit dcaf93b7d9
1 changed files with 14 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,6 +19,7 @@ package org.springframework.integration.mqtt.inbound;
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 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
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 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
}
}
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 class Mqttv5PahoMessageDrivenChannelAdapter
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)) {