diff options
| -rw-r--r-- | kafka/coordinator/base.py | 8 |
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 |