diff --git a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
index aa9076037a..b77a1943a7 100644
--- a/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
+++ b/spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java
@@ -43,6 +43,8 @@ import org.springframework.util.Assert;
  * Eclipse Paho Implementation.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 1.0
  *
  */
@@ -172,6 +174,9 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 			catch (MqttException e) {
 				logger.error("Exception while disconnecting", e);
 			}
+
+			this.client.setCallback(null);
+
 			try {
 				this.client.close();
 			}
@@ -308,11 +313,13 @@ public class MqttPahoMessageDrivenChannelAdapter extends AbstractMqttMessageDriv
 
 	@Override
 	public synchronized void connectionLost(Throwable cause) {
-		this.logger.error("Lost connection:" + cause.getMessage() + "; retrying...");
-		this.connected = false;
-		scheduleReconnect();
-		if (this.applicationEventPublisher != null) {
-			this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));
+		if (isRunning()) {
+			this.logger.error("Lost connection: " + cause.getMessage() + "; retrying...");
+			this.connected = false;
+			scheduleReconnect();
+			if (this.applicationEventPublisher != null) {
+				this.applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, cause));
+			}
 		}
 	}
 
diff --git a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
index d4590f1c08..3a66b4a089 100644
--- a/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
+++ b/spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java
@@ -40,6 +40,7 @@ import static org.mockito.Mockito.verify;
 
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.Date;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -62,7 +63,6 @@ import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
-import org.eclipse.paho.client.mqttv3.MqttSecurityException;
 import org.eclipse.paho.client.mqttv3.MqttToken;
 import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import org.junit.Test;
@@ -97,6 +97,7 @@ import org.springframework.util.ReflectionUtils;
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
@@ -345,6 +346,13 @@ public class MqttAdapterTests {
 		adapter.start();
 		adapter.stop();
 		verifyUnsubscribe(client);
+
+		adapter.connectionLost(new RuntimeException("Intentional"));
+
+		TaskScheduler taskScheduler = TestUtils.getPropertyValue(adapter, "taskScheduler", TaskScheduler.class);
+
+		verify(taskScheduler, never())
+				.schedule(any(Runnable.class), any(Date.class));
 	}
 
 	@Test
@@ -509,7 +517,7 @@ public class MqttAdapterTests {
 	}
 
 	private MqttPahoMessageDrivenChannelAdapter buildAdapter(final IMqttClient client, Boolean cleanSession,
-			ConsumerStopAction action) throws MqttException, MqttSecurityException {
+			ConsumerStopAction action) throws MqttException {
 		DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory() {
 
 			@Override