GH-2471: MQTT: Fix Thread Leak

Fixes https://github.com/spring-projects/spring-integration/issues/2471

Call `close()` on the client whenever the connection is lost or can't be
established, to release resources in the client.

**cherry-pick to 5.0.x, 4.3.x**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -262,6 +262,14 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 			}
 			logger.error("Error connecting or subscribing to " + Arrays.toString(topics), e);
 			this.client.disconnectForcibly(this.completionTimeout);
+			try {
+				this.client.setCallback(null);
+				this.client.close();
+			}
+			catch (MqttException e1) {
+				// NOSONAR
+			}
+			this.client = null;
 			throw e;
 		}
 		finally {
@@ -316,6 +324,14 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 		if (isRunning()) {
 			this.logger.error("Lost connection: " + cause.getMessage() + "; retrying...");
 			this.connected = false;
+			try {
+				this.client.setCallback(null);
+				this.client.close();
+			}
+			catch (MqttException e) {
+				// NOSONAR
+			}
+			this.client = null;
 			scheduleReconnect();
 			if (this.applicationEventPublisher != null) {
 				this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));

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

@@ -31,6 +31,7 @@ import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.BDDMockito.willThrow;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -111,6 +112,16 @@ public class MqttAdapterTests {
 		this.alwaysComplete = (IMqttToken) pfb.getObject();
 	}
 
+	@Test
+	public void testCloseOnBadConnect() throws Exception {
+		final IMqttClient client = mock(IMqttClient.class);
+		willThrow(new MqttException(0)).given(client).connect(any());
+		MqttPahoMessageDrivenChannelAdapter adapter = buildAdapter(client, null, ConsumerStopAction.UNSUBSCRIBE_NEVER);
+		adapter.start();
+		verify(client).close();
+		adapter.stop();
+	}
+
 	@Test
 	public void testOutboundOptionsApplied() throws Exception {
 		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
@@ -383,6 +394,7 @@ public class MqttAdapterTests {
 		adapter.setTaskScheduler(taskScheduler);
 		adapter.start();
 		adapter.connectionLost(new RuntimeException("initial"));
+		verify(client).close();
 		Thread.sleep(1000);
 		// the following assertion should be equalTo, but leq to protect against a slow CI server
 		assertThat(attemptingReconnectCount.get(), lessThanOrEqualTo(2));