openstack/deb-python-kafka
Code Issues Proposed changes

Log successful heartbeat as INFO; improve heartbeat response logging

Browse Source
This commit is contained in:
Dana Powers
2016-03-11 16:47:46 -08:00
parent 0057e75ecf
commit 86d98c00fd
1 changed files with 12 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
kafka/coordinator/base.py
View File

@@ -536,26 +536,27 @@ class BaseCoordinator(object):
#self.sensors.heartbeat_latency.record(response.requestLatencyMs())
error_type = Errors.for_code(response.error_code)
if error_type is Errors.NoError:
log.debug("Received successful heartbeat response.")
log.info("Heartbeat successful")
future.success(None)
elif error_type in (Errors.GroupCoordinatorNotAvailableError,
Errors.NotCoordinatorForGroupError):
log.info("Heartbeat failed: coordinator is either not started or"
" not valid; will refresh metadata and retry")
log.warning("Heartbeat failed: coordinator is either not started or"
" not valid; will refresh metadata and retry")
self.coordinator_dead()
future.failure(error_type())
elif error_type is Errors.RebalanceInProgressError:
log.info("Heartbeat failed: group is rebalancing; re-joining group")
log.warning("Heartbeat: group is rebalancing; this consumer needs to"
" re-join")
self.rejoin_needed = True
future.failure(error_type())
elif error_type is Errors.IllegalGenerationError:
log.info("Heartbeat failed: local generation id is not current;"
" re-joining group")
log.warning("Heartbeat: generation id is not current; this consumer"
" needs to re-join")
self.rejoin_needed = True
future.failure(error_type())
elif error_type is Errors.UnknownMemberIdError:
log.info("Heartbeat failed: local member_id was not recognized;"
" resetting and re-joining group")
log.warning("Heartbeat: local member_id was not recognized;"
" this consumer needs to re-join")
self.member_id = JoinGroupRequest.UNKNOWN_MEMBER_ID
self.rejoin_needed = True
future.failure(error_type)
@@ -597,8 +598,8 @@ class HeartbeatTask(object):
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")
log.info("Skipping heartbeat: no auto-assignment"
" or waiting on rebalance")
return
if self._coordinator.coordinator_unknown():
@@ -633,7 +634,7 @@ class HeartbeatTask(object):
self._client.schedule(self, time.time() + ttl)
def _handle_heartbeat_failure(self, e):
log.debug("Heartbeat failed; retrying")
log.warning("Heartbeat failed; retrying")
self._request_in_flight = False
etd = time.time() + self._coordinator.config['retry_backoff_ms'] / 1000.0
self._client.schedule(self, etd)