From 23c1d5a3e798c811292bfd37d3812a6fc68cab19 Mon Sep 17 00:00:00 2001 From: Jeff Widman Date: Fri, 6 Oct 2017 16:31:51 -0700 Subject: [PATCH 1/4] Expand metrics docstrings --- kafka/consumer/group.py | 9 +++++++-- kafka/producer/kafka.py | 15 ++++++++++----- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index b7fbd8395..049d5d9e8 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -846,8 +846,13 @@ def unsubscribe(self): log.debug("Unsubscribed all topics or patterns and assigned partitions") def metrics(self, raw=False): - """Warning: this is an unstable interface. - It may change in future releases without warning""" + """Get metrics on consumer performance. + + This is ported from the Java Consumer, for details see: + https://kafka.apache.org/documentation/#new_consumer_monitoring + + Warning: This is an unstable interface. It may change in future + releases without warning.""" if raw: return self._metrics.metrics diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 09ca74401..18a313960 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -566,10 +566,10 @@ def flush(self, timeout=None): Arguments: timeout (float, optional): timeout in seconds to wait for completion. - + Raises: - KafkaTimeoutError: failure to flush buffered records within the - provided timeout + KafkaTimeoutError: failure to flush buffered records within the + provided timeout """ log.debug("Flushing accumulated records in producer.") # trace self._accumulator.begin_flush() @@ -655,8 +655,13 @@ def _partition(self, topic, partition, key, value, available) def metrics(self, raw=False): - """Warning: this is an unstable interface. - It may change in future releases without warning""" + """Get metrics on producer performance. + + This is ported from the Java Producer, for details see: + https://kafka.apache.org/documentation/#producer_monitoring + + Warning: This is an unstable interface. It may change in future + releases without warning.""" if raw: return self._metrics.metrics From 23e6a1afdaff57fbe8134197eebfaa95628f15f5 Mon Sep 17 00:00:00 2001 From: Jeff Widman Date: Fri, 6 Oct 2017 16:34:20 -0700 Subject: [PATCH 2/4] Document metrics interface in readme --- README.rst | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/README.rst b/README.rst index 6e9a50714..d4fc1a9ad 100644 --- a/README.rst +++ b/README.rst @@ -70,6 +70,8 @@ that expose basic message attributes: topic, partition, offset, key, and value: >>> for msg in consumer: ... assert isinstance(msg.value, dict) +>>> # Get consumer metrics +>>> metrics = consumer.metrics() KafkaProducer ************* @@ -110,6 +112,9 @@ for more details. >>> for i in range(1000): ... producer.send('foobar', b'msg %d' % i) +>>> # Get producer performance metrics +>>> metrics = producer.metrics() + Thread safety ************* @@ -122,8 +127,8 @@ multiprocessing is recommended. Compression *********** -kafka-python supports gzip compression/decompression natively. To produce or consume lz4 -compressed messages, you should install python-lz4 (pip install lz4). +kafka-python supports gzip compression/decompression natively. To produce or consume lz4 +compressed messages, you should install python-lz4 (pip install lz4). To enable snappy compression/decompression install python-snappy (also requires snappy library). See for more information. @@ -138,7 +143,6 @@ leveraged to enable a KafkaClient.check_version() method that probes a kafka broker and attempts to identify which version it is running (0.8.0 to 0.11). - Low-level ********* From 0e54b123a29c630bae6225bf34e25d4bd2bc6997 Mon Sep 17 00:00:00 2001 From: Jeff Widman Date: Fri, 6 Oct 2017 16:35:03 -0700 Subject: [PATCH 3/4] Use six.iteritems(d) rather than d.items() --- kafka/consumer/group.py | 4 ++-- kafka/producer/kafka.py | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 049d5d9e8..8ae7be0ea 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -857,7 +857,7 @@ def metrics(self, raw=False): return self._metrics.metrics metrics = {} - for k, v in self._metrics.metrics.items(): + for k, v in six.iteritems(self._metrics.metrics): if k.group not in metrics: metrics[k.group] = {} if k.name not in metrics[k.group]: @@ -902,7 +902,7 @@ def offsets_for_times(self, timestamps): raise UnsupportedVersionError( "offsets_for_times API not supported for cluster version {}" .format(self.config['api_version'])) - for tp, ts in timestamps.items(): + for tp, ts in six.iteritems(timestamps): timestamps[tp] = int(ts) if ts < 0: raise ValueError( diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 18a313960..2a325ca47 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -8,6 +8,8 @@ import time import weakref +from ..vendor import six + from .. import errors as Errors from ..client_async import KafkaClient, selectors from ..metrics import MetricConfig, Metrics @@ -666,7 +668,7 @@ def metrics(self, raw=False): return self._metrics.metrics metrics = {} - for k, v in self._metrics.metrics.items(): + for k, v in six.iteritems(self._metrics.metrics): if k.group not in metrics: metrics[k.group] = {} if k.name not in metrics[k.group]: From ad3d8708e9597ae5ceee26796fe60b7f22bfb08e Mon Sep 17 00:00:00 2001 From: Jeff Widman Date: Fri, 6 Oct 2017 16:36:57 -0700 Subject: [PATCH 4/4] Use Sphinx warning syntax --- kafka/consumer/group.py | 6 ++++-- kafka/producer/kafka.py | 6 ++++-- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 8ae7be0ea..a83d5da6e 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -851,8 +851,10 @@ def metrics(self, raw=False): This is ported from the Java Consumer, for details see: https://kafka.apache.org/documentation/#new_consumer_monitoring - Warning: This is an unstable interface. It may change in future - releases without warning.""" + Warning: + This is an unstable interface. It may change in future + releases without warning. + """ if raw: return self._metrics.metrics diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 2a325ca47..de9dcd2ce 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -662,8 +662,10 @@ def metrics(self, raw=False): This is ported from the Java Producer, for details see: https://kafka.apache.org/documentation/#producer_monitoring - Warning: This is an unstable interface. It may change in future - releases without warning.""" + Warning: + This is an unstable interface. It may change in future + releases without warning. + """ if raw: return self._metrics.metrics