Avoid self-join in heartbeat thread close (#2488)

Author: dpkp

kafka/coordinator/base.py

@@ -920,9 +920,18 @@ def disable(self):
         self.enabled = False
 
     def close(self):
+        if self.closed:
+            return
         self.closed = True
         with self.coordinator._lock:
             self.coordinator._lock.notify()
+
+        # Generally this should not happen - close() is triggered
+        # by the coordinator. But in some cases GC may close the coordinator
+        # from within the heartbeat thread.
+        if threading.current_thread() == self:
+            return
+
         if self.is_alive():
             self.join(self.coordinator.config['heartbeat_interval_ms'] / 1000)
         if self.is_alive():