diff --git a/oslo_messaging/_drivers/impl_rabbit.py b/oslo_messaging/_drivers/impl_rabbit.py
index 8c696cb7e..75041cd8a 100644
--- a/oslo_messaging/_drivers/impl_rabbit.py
+++ b/oslo_messaging/_drivers/impl_rabbit.py
@@ -231,6 +231,9 @@ class Consumer(object):
                            consumer_tag=six.text_type(tag),
                            nowait=self.nowait)
 
+    def cancel(self, tag):
+        self.queue.cancel(six.text_type(tag))
+
     def _callback(self, message):
         """Call callback with deserialized message.
 
@@ -689,11 +692,12 @@ class Connection(object):
 
         with self._connection_lock:
             try:
-                self._set_current_channel(self.connection.channel())
+                for tag, consumer in enumerate(self._consumers):
+                    consumer.cancel(tag=tag)
             except recoverable_errors:
                 self._set_current_channel(None)
                 self.ensure_connection()
-        self._consumers = []
+            self._consumers = []
 
     def _heartbeat_supported_and_enabled(self):
         if self.driver_conf.heartbeat_timeout_threshold <= 0:
diff --git a/oslo_messaging/tests/drivers/test_impl_rabbit.py b/oslo_messaging/tests/drivers/test_impl_rabbit.py
index 1e942a9cf..48ffffca6 100644
--- a/oslo_messaging/tests/drivers/test_impl_rabbit.py
+++ b/oslo_messaging/tests/drivers/test_impl_rabbit.py
@@ -257,18 +257,6 @@ class TestRabbitConsume(test_utils.BaseTestCase):
 
         self.assertEqual(0, int(deadline - time.time()))
 
-    def test_connection_reset_always_succeed(self):
-        transport = oslo_messaging.get_transport(self.conf,
-                                                 'kombu+memory:////')
-        self.addCleanup(transport.cleanup)
-        channel = mock.Mock()
-        with transport._driver._get_connection(amqp.PURPOSE_LISTEN) as conn:
-            conn.connection.connection.recoverable_channel_errors = (IOError,)
-            with mock.patch.object(conn.connection.connection, 'channel',
-                                   side_effect=[IOError, IOError, channel]):
-                conn.connection.reset()
-                self.assertEqual(channel, conn.connection.channel)
-
     def test_connection_ack_have_disconnected_kombu_connection(self):
         transport = oslo_messaging.get_transport(self.conf,
                                                  'kombu+memory:////')