Expand metrics docs (#1243)

* Expand metrics docstrings
* Document metrics interface in readme
* Use six.iteritems(d) rather than d.items()
* Use Sphinx warning syntax

Author: jeffwidman

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 <https://kafka-python.readthedocs.io/en/master/install.html#optional-snappy-install>
 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
 *********
 

kafka/consumer/group.py

@@ -846,13 +846,20 @@ 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
 
         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]:
@@ -897,7 +904,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(

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
@@ -566,10 +568,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,13 +657,20 @@ 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
 
         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]: