diff options
Diffstat (limited to 'kafka/consumer/fetcher.py')
| -rw-r--r-- | kafka/consumer/fetcher.py | 8 |
1 files changed, 6 insertions, 2 deletions
diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 34ff4cb..d615848 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -729,6 +729,8 @@ class Fetcher(six.Iterator): else: raise error_type('Unexpected error while fetching data') + # Because we are currently decompressing messages lazily, the sensors here + # will get compressed bytes / message set stats when compression is enabled self._sensors.bytes_fetched.record(total_bytes) self._sensors.records_fetched.record(total_count) if response.API_VERSION >= 1: @@ -774,12 +776,12 @@ class FetchManagerMetrics(object): 'The maximum throttle time in ms'), Max()) def record_topic_fetch_metrics(self, topic, num_bytes, num_records): - metric_tags = {'topic': topic.replace('.', '_')} - # record bytes fetched name = '.'.join(['topic', topic, 'bytes-fetched']) bytes_fetched = self.metrics.get_sensor(name) if not bytes_fetched: + metric_tags = {'topic': topic.replace('.', '_')} + bytes_fetched = self.metrics.sensor(name) bytes_fetched.add(self.metrics.metric_name('fetch-size-avg', self.group_name, @@ -799,6 +801,8 @@ class FetchManagerMetrics(object): name = '.'.join(['topic', topic, 'records-fetched']) records_fetched = self.metrics.get_sensor(name) if not records_fetched: + metric_tags = {'topic': topic.replace('.', '_')} + records_fetched = self.metrics.sensor(name) records_fetched.add(self.metrics.metric_name('records-per-request-avg', self.group_name, |