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 3adcd19961..c63e229126 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();
 			}
@@ -313,11 +318,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 e054d1cfdd..ac2b5b1bc6 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
@@ -26,15 +26,16 @@ import static org.junit.Assert.assertTrue;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.willAnswer;
 import static org.mockito.BDDMockito.willReturn;
-import static org.mockito.Mockito.any;
-import static org.mockito.Mockito.anyLong;
-import static org.mockito.Mockito.anyString;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyLong;
+import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
+import java.util.Date;
 import java.util.Properties;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
@@ -56,7 +57,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;
@@ -83,6 +83,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.0
  *
  */
@@ -331,6 +333,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
@@ -372,7 +381,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