Update _close_connection to ensure connection is closed before joinin…

…g the thread

Remove unused import per review

neon_mq_connector/consumers/select_consumer.py

@@ -30,7 +30,7 @@
 import threading
 import time
 
-from asyncio import Event, Lock, run
+from asyncio import Event
 from typing import Optional
 
 import pika.exceptions
@@ -233,6 +233,13 @@ def _close_connection(self, mark_consumer_as_dead: bool = True):
                     raise TimeoutError(f"Timeout waiting for channel close. "
                                        f"is_closed={self.channel.is_closed}")
                 LOG.info(f"Channel closed")
+
+                # Wait for the connection to close
+                waiter = threading.Event()
+                while not self.connection.is_closed:
+                    waiter.wait(1)
+                LOG.info(f"Connection closed")
+
             if self.connection:
                 self.connection.ioloop.stop()
             # self.connection = None