HeartbeatTask should reschedule heartbeat on coordinator_unknown()

1 files changed, 6 insertions, 2 deletions

diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index c49c38b..a2c47a4 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -594,14 +594,18 @@ class HeartbeatTask(object):
 
     def __call__(self):
         if (self._coordinator.generation < 0 or
-            self._coordinator.need_rejoin() or
-            self._coordinator.coordinator_unknown()):
+            self._coordinator.need_rejoin()):
             # no need to send the heartbeat we're not using auto-assignment
             # or if we are awaiting a rebalance
             log.debug("Skipping heartbeat: no auto-assignment"
                       " or waiting on rebalance")
             return
 
+        if self._coordinator.coordinator_unknown():
+            log.warning("Coordinator unknown during heartbeat -- will retry")
+            self._handle_heartbeat_failure(Errors.GroupCoordinatorNotAvailableError())
+            return
+
         if self._heartbeat.session_expired():
             # we haven't received a successful heartbeat in one session interval
             # so mark the coordinator dead