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

GH-3627: Fix race condition NPE in MqttPahoMDCA

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3627

The `destroy()`, and therefore `stop()` could be called from the `MqttConnectionFailedEvent` handling
in the same thread resetting the `client` property to `null`.

* Check for `this.client != null` in the next block of the `connectAndSubscribe()` to avoid NPE
* Check for `isActive()` in the `scheduleReconnect()` to be sure do not reconnect if channel adapter
has been stopped already

**Cherry-pick to `5.4.x`**
This commit is contained in:
Artem Bilan
2021-09-09 14:17:57 -04:00
committed by Gary Russell
parent 9f426500a1
commit 8ff7ad798d
2 changed files with 80 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
View File

@@ -84,6 +84,8 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private boolean manualAcks;
private ApplicationEventPublisher applicationEventPublisher;
private volatile IMqttClient client;
private volatile ScheduledFuture<?> reconnectFuture;
@@ -94,8 +96,6 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private volatile ConsumerStopAction consumerStopAction;
private ApplicationEventPublisher applicationEventPublisher;
/**
* Use this constructor for a single url (although it may be overridden if the server
* URI(s) are provided by the {@link MqttConnectOptions#getServerURIs()} provided by
@@ -311,15 +311,17 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex));
}
logger.error(ex, () -> "Error connecting or subscribing to " + Arrays.toString(topics));
this.client.disconnectForcibly(this.disconnectCompletionTimeout);
try {
this.client.setCallback(null);
this.client.close();
if (this.client != null) { // Could be reset during event handling before
this.client.disconnectForcibly(this.disconnectCompletionTimeout);
try {
this.client.setCallback(null);
this.client.close();
}
catch (MqttException e1) {
// NOSONAR
}
this.client = null;
}
catch (MqttException e1) {
// NOSONAR
}
this.client = null;
throw ex;
}
finally {
@@ -355,25 +357,27 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
private synchronized void scheduleReconnect() {
cancelReconnect();
try {
this.reconnectFuture = getTaskScheduler().schedule(() -> {
try {
logger.debug("Attempting reconnect");
synchronized (MqttPahoMessageDrivenChannelAdapter.this) {
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
MqttPahoMessageDrivenChannelAdapter.this.reconnectFuture = null;
if (isActive()) {
try {
this.reconnectFuture = getTaskScheduler().schedule(() -> {
try {
logger.debug("Attempting reconnect");
synchronized (MqttPahoMessageDrivenChannelAdapter.this) {
if (!MqttPahoMessageDrivenChannelAdapter.this.connected) {
connectAndSubscribe();
MqttPahoMessageDrivenChannelAdapter.this.reconnectFuture = null;
}
}
}
}
catch (MqttException ex) {
logger.error(ex, "Exception while connecting and subscribing");
scheduleReconnect();
}
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
}
catch (Exception ex) {
logger.error(ex, "Failed to schedule reconnect");
catch (MqttException ex) {
logger.error(ex, "Exception while connecting and subscribing");
scheduleReconnect();
}
}, new Date(System.currentTimeMillis() + this.recoveryInterval));
}
catch (Exception ex) {
logger.error(ex, "Failed to schedule reconnect");
}
}
}
@@ -412,7 +416,7 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
sendMessage(message);
}
catch (RuntimeException ex) {
logger.error(ex, () -> "Unhandled exception for " + message.toString());
logger.error(ex, () -> "Unhandled exception for " + message);
throw ex;
}
}

53
spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
View File

@@ -22,6 +22,7 @@ import static org.assertj.core.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
@@ -53,7 +54,6 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.assertj.core.api.Condition;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttClient;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
@@ -65,6 +65,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttToken;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.ArgumentMatchers;
import org.mockito.internal.stubbing.answers.CallsRealMethods;
@@ -540,8 +541,7 @@ public class MqttAdapterTests {
new DirectFieldAccessor(client).setPropertyValue("aClient", aClient);
willAnswer(new CallsRealMethods()).given(client).connect(any(MqttConnectOptions.class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class));
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class),
(IMqttMessageListener[]) isNull());
willAnswer(new CallsRealMethods()).given(client).subscribe(any(String[].class), any(int[].class), isNull());
willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
IMqttToken token = mock(IMqttToken.class);
@@ -572,8 +572,51 @@ public class MqttAdapterTests {
verify(client).disconnectForcibly(5_000L);
}
@Test
public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
final IMqttClient client = mock(IMqttClient.class);
MqttPahoMessageDrivenChannelAdapter adapter = buildAdapterIn(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
adapter.setRecoveryInterval(10);
MqttException mqttException = new MqttException(MqttException.REASON_CODE_SUBSCRIBE_FAILED);
willThrow(mqttException)
.given(client)
.subscribe(any(), ArgumentMatchers.<int[]>any());
LogAccessor logger = spy(TestUtils.getPropertyValue(adapter, "logger", LogAccessor.class));
new DirectFieldAccessor(adapter).setPropertyValue("logger", logger);
CountDownLatch exceptionLatch = new CountDownLatch(1);
ArgumentCaptor<MqttException> mqttExceptionArgumentCaptor = ArgumentCaptor.forClass(MqttException.class);
willAnswer(i -> {
exceptionLatch.countDown();
return null;
})
.given(logger)
.error(mqttExceptionArgumentCaptor.capture(), eq("Exception while connecting and subscribing"));
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
taskScheduler.initialize();
adapter.setTaskScheduler(taskScheduler);
adapter.setApplicationEventPublisher(event -> {
if (event instanceof MqttConnectionFailedEvent) {
adapter.destroy();
}
});
adapter.start();
assertThat(exceptionLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(mqttExceptionArgumentCaptor.getValue())
.isNotNull()
.isSameAs(mqttException);
taskScheduler.destroy();
}
private MqttPahoMessageDrivenChannelAdapter buildAdapterIn(final IMqttClient client, Boolean cleanSession,
ConsumerStopAction action) throws MqttException {
ConsumerStopAction action) {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override
@@ -604,7 +647,7 @@ public class MqttAdapterTests {
DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
@Override
public IMqttAsyncClient getAsyncClientInstance(String uri, String clientId) throws MqttException {
public IMqttAsyncClient getAsyncClientInstance(String uri, String clientId) {
return client;
}