Check for immediate failure when looking up coordinator in heartbeat …

…thread (dpkp#1457)

kafka/coordinator/base.py

@@ -940,21 +940,17 @@ def _run_once(self):
                 self.disable()
                 return
 
-        # TODO: When consumer.wakeup() is implemented, we need to
-        # disable here to prevent propagating an exception to this
-        # heartbeat thread
-        #
-        # Release coordinator lock during client poll to avoid deadlocks
-        # if/when connection errback needs coordinator lock
-        self.coordinator._client.poll(timeout_ms=0)
-
-        if self.coordinator.coordinator_unknown():
-            future = self.coordinator.lookup_coordinator()
-            if not future.is_done or future.failed():
-                # the immediate future check ensures that we backoff
-                # properly in the case that no brokers are available
-                # to connect to (and the future is automatically failed).
-                with self.coordinator._lock:
+            # TODO: When consumer.wakeup() is implemented, we need to
+            # disable here to prevent propagating an exception to this
+            # heartbeat thread
+            self.coordinator._client.poll(timeout_ms=0)
+
+            if self.coordinator.coordinator_unknown():
+                future = self.coordinator.lookup_coordinator()
+                if not future.is_done or future.failed():
+                    # the immediate future check ensures that we backoff
+                    # properly in the case that no brokers are available
+                    # to connect to (and the future is automatically failed).
                     self.coordinator._lock.wait(self.coordinator.config['retry_backoff_ms'] / 1000)
 
         elif self.coordinator.heartbeat.session_timeout_expired():