openstack/deb-python-kafka
Code Issues Proposed changes

HeartbeatTask should reschedule heartbeat on coordinator_unknown()

Browse Source
This commit is contained in:
Dana Powers
2016-03-11 16:41:42 -08:00
parent f58b7e9b2a
commit 0057e75ecf
1 changed files with 6 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
kafka/coordinator/base.py
View File

@@ -594,14 +594,18 @@ class HeartbeatTask(object):
def __call__(self): def __call__(self):
if (self._coordinator.generation < 0 or if (self._coordinator.generation < 0 or
self._coordinator.need_rejoin() or self._coordinator.need_rejoin()):
self._coordinator.coordinator_unknown()):
# no need to send the heartbeat we're not using auto-assignment # no need to send the heartbeat we're not using auto-assignment
# or if we are awaiting a rebalance # or if we are awaiting a rebalance
log.debug("Skipping heartbeat: no auto-assignment" log.debug("Skipping heartbeat: no auto-assignment"
" or waiting on rebalance") " or waiting on rebalance")
return 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(): if self._heartbeat.session_expired():
# we haven't received a successful heartbeat in one session interval # we haven't received a successful heartbeat in one session interval
# so mark the coordinator dead # so mark the coordinator dead