From 0057e75ecfff12dcc16ec5b285c7288666798552 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Fri, 11 Mar 2016 16:41:42 -0800 Subject: HeartbeatTask should reschedule heartbeat on coordinator_unknown() --- kafka/coordinator/base.py | 8 ++++++-- 1 file 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 -- cgit v1.2.1