Merge branch 'master' into master

Author: Tincu Gabriel

.travis.yml

@@ -17,6 +17,7 @@ env:
     - KAFKA_VERSION=1.1.1
     - KAFKA_VERSION=2.4.0
     - KAFKA_VERSION=2.5.0
+    - KAFKA_VERSION=2.6.0
 
 addons:
   apt:

README.rst

@@ -1,7 +1,7 @@
 Kafka Python client
 ------------------------
 
-.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/en/master/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
@@ -34,6 +34,7 @@ documentation, please see readthedocs and/or python's inline help.
 
 >>> pip install kafka-python
 
+
 KafkaConsumer
 *************
 
@@ -78,6 +79,7 @@ that expose basic message attributes: topic, partition, offset, key, and value:
 >>> # Get consumer metrics
 >>> metrics = consumer.metrics()
 
+
 KafkaProducer
 *************
 
@@ -124,6 +126,7 @@ for more details.
 >>> # Get producer performance metrics
 >>> metrics = producer.metrics()
 
+
 Thread safety
 *************
 
@@ -133,22 +136,30 @@ KafkaConsumer which cannot.
 While it is possible to use the KafkaConsumer in a thread-local manner,
 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).
-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.
+kafka-python supports the following compression formats:
+
+- gzip
+- LZ4
+- Snappy
+- Zstandard (zstd)
+
+gzip is supported natively, the others require installing additional libraries.
+See <https://kafka-python.readthedocs.io/en/master/install.html> for more information.
+
 
 Optimized CRC32 Validation
 **************************
 
 Kafka uses CRC32 checksums to validate messages. kafka-python includes a pure
 python implementation for compatibility. To improve performance for high-throughput
 applications, kafka-python will use `crc32c` for optimized native code if installed.
-See https://pypi.org/project/crc32c/
+See <https://kafka-python.readthedocs.io/en/master/install.html> for installation instructions.
+See https://pypi.org/project/crc32c/ for details on the underlying crc32c lib.
+
 
 Protocol
 ********
@@ -158,4 +169,4 @@ for interacting with kafka brokers via the python repl. This is useful for
 testing, probing, and general experimentation. The protocol support is
 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 2.4+).
+(0.8.0 to 2.6+).

docs/compatibility.rst

@@ -1,12 +1,12 @@
 Compatibility
 -------------
 
-.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
 
-kafka-python is compatible with (and tested against) broker versions 2.5
+kafka-python is compatible with (and tested against) broker versions 2.6
 through 0.8.0 . kafka-python is not compatible with the 0.8.2-beta release.
 
 Because the kafka server protocol is backwards compatible, kafka-python is

docs/index.rst

@@ -1,7 +1,7 @@
 kafka-python
 ############
 
-.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
+.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
     :target: https://kafka-python.readthedocs.io/compatibility.html
 .. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
     :target: https://pypi.python.org/pypi/kafka-python
@@ -33,6 +33,7 @@ documentation, please see readthedocs and/or python's inline help.
 
 >>> pip install kafka-python
 
+
 KafkaConsumer
 *************
 
@@ -122,12 +123,26 @@ multiprocessing is recommended.
 Compression
 ***********
 
-kafka-python supports multiple compression types:
+kafka-python supports the following compression formats:
+
+ - gzip
+ - LZ4
+ - Snappy
+ - Zstandard (zstd)
+
+gzip is supported natively, the others require installing additional libraries.
+See `Install <install.html>`_ for more information.
+
+
+Optimized CRC32 Validation
+**************************
+
+Kafka uses CRC32 checksums to validate messages. kafka-python includes a pure
+python implementation for compatibility. To improve performance for high-throughput
+applications, kafka-python will use `crc32c` for optimized native code if installed.
+See `Install <install.html>`_ for installation instructions and
+https://pypi.org/project/crc32c/ for details on the underlying crc32c lib.
 
- - gzip : supported natively
- - lz4 : requires `python-lz4 <https://pypi.org/project/lz4/>`_ installed
- - snappy : requires the `python-snappy <https://pypi.org/project/python-snappy/>`_  package (which requires the snappy C library)
- - zstd : requires the `python-zstandard <https://github.com/indygreg/python-zstandard>`_ package installed
 
 Protocol
 ********
@@ -137,7 +152,7 @@ for interacting with kafka brokers via the python repl. This is useful for
 testing, probing, and general experimentation. The protocol support is
 leveraged to enable a :meth:`~kafka.KafkaClient.check_version()`
 method that probes a kafka broker and
-attempts to identify which version it is running (0.8.0 to 2.4+).
+attempts to identify which version it is running (0.8.0 to 2.6+).
 
 
 .. toctree::

docs/install.rst

@@ -23,20 +23,33 @@ Bleeding-Edge
     pip install ./kafka-python
 
 
-Optional LZ4 install
+Optional crc32c install
+***********************
+Highly recommended if you are using Kafka 11+ brokers. For those `kafka-python`
+uses a new message protocol version, that requires calculation of `crc32c`,
+which differs from the `zlib.crc32` hash implementation. By default `kafka-python`
+calculates it in pure python, which is quite slow. To speed it up we optionally
+support https://pypi.python.org/pypi/crc32c package if it's installed.
+
+.. code:: bash
+
+    pip install 'kafka-python[crc32c]'
+
+
+Optional ZSTD install
 ********************
 
-To enable LZ4 compression/decompression, install python-lz4:
+To enable ZSTD compression/decompression, install python-zstandard:
 
->>> pip install lz4
+>>> pip install 'kafka-python[zstd]'
 
 
-Optional crc32c install
+Optional LZ4 install
 ********************
 
-To enable optimized CRC32 checksum validation, install crc32c:
+To enable LZ4 compression/decompression, install python-lz4:
 
->>> pip install crc32c
+>>> pip install 'kafka-python[lz4]'
 
 
 Optional Snappy install
@@ -77,17 +90,4 @@ Install the `python-snappy` module
 
 .. code:: bash
 
-    pip install python-snappy
-
-
-Optional crc32c install
-***********************
-Highly recommended if you are using Kafka 11+ brokers. For those `kafka-python`
-uses a new message protocol version, that requires calculation of `crc32c`,
-which differs from `zlib.crc32` hash implementation. By default `kafka-python`
-calculates it in pure python, which is quite slow. To speed it up we optionally
-support https://pypi.python.org/pypi/crc32c package if it's installed.
-
-.. code:: bash
-
-    pip install crc32c
+    pip install 'kafka-python[snappy]'

kafka/conn.py

@@ -24,7 +24,7 @@
 from kafka.future import Future
 from kafka.metrics.stats import Avg, Count, Max, Rate
 from kafka.oauth.abstract import AbstractTokenProvider
-from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2
+from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2, DescribeClientQuotasRequest
 from kafka.protocol.commit import OffsetFetchRequest
 from kafka.protocol.offset import OffsetRequest
 from kafka.protocol.produce import ProduceRequest
@@ -1169,6 +1169,7 @@ def _infer_broker_version_from_api_versions(self, api_versions):
         # in reverse order. As soon as we find one that works, return it
         test_cases = [
             # format (<broker version>, <needed struct>)
+            ((2, 6, 0), DescribeClientQuotasRequest[0]),
             ((2, 5, 0), DescribeAclsRequest_v2),
             ((2, 4, 0), ProduceRequest[8]),
             ((2, 3, 0), FetchRequest[11]),

kafka/consumer/fetcher.py

@@ -817,8 +817,9 @@ def _parse_fetched_data(self, completed_fetch):
                               position)
                     unpacked = list(self._unpack_message_set(tp, records))
                     parsed_records = self.PartitionRecords(fetch_offset, tp, unpacked)
-                    last_offset = unpacked[-1].offset
-                    self._sensors.records_fetch_lag.record(highwater - last_offset)
+                    if unpacked:
+                        last_offset = unpacked[-1].offset
+                        self._sensors.records_fetch_lag.record(highwater - last_offset)
                     num_bytes = records.valid_bytes()
                     records_count = len(unpacked)
                 elif records.size_in_bytes() > 0:

kafka/consumer/group.py

@@ -651,7 +651,7 @@ def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
         # Poll for new data until the timeout expires
         start = time.time()
         remaining = timeout_ms
-        while True:
+        while not self._closed:
             records = self._poll_once(remaining, max_records, update_offsets=update_offsets)
             if records:
                 return records
@@ -660,7 +660,9 @@ def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
             remaining = timeout_ms - elapsed_ms
 
             if remaining <= 0:
-                return {}
+                break
+
+        return {}
 
     def _poll_once(self, timeout_ms, max_records, update_offsets=True):
         """Do one round of polling. In addition to checking for new data, this does

kafka/coordinator/assignors/sticky/sticky_assignor.py

@@ -648,15 +648,19 @@ def parse_member_metadata(cls, metadata):
 
     @classmethod
     def metadata(cls, topics):
-        if cls.member_assignment is None:
+        return cls._metadata(topics, cls.member_assignment, cls.generation)
+
+    @classmethod
+    def _metadata(cls, topics, member_assignment_partitions, generation=-1):
+        if member_assignment_partitions is None:
             log.debug("No member assignment available")
             user_data = b''
         else:
             log.debug("Member assignment is available, generating the metadata: generation {}".format(cls.generation))
             partitions_by_topic = defaultdict(list)
-            for topic_partition in cls.member_assignment:   # pylint: disable=not-an-iterable
+            for topic_partition in member_assignment_partitions:
                 partitions_by_topic[topic_partition.topic].append(topic_partition.partition)
-            data = StickyAssignorUserDataV1(six.iteritems(partitions_by_topic), cls.generation)
+            data = StickyAssignorUserDataV1(six.viewitems(partitions_by_topic), generation)
             user_data = data.encode()
         return ConsumerProtocolMemberMetadata(cls.version, list(topics), user_data)
 

kafka/protocol/__init__.py

@@ -45,4 +45,5 @@
     42: 'DeleteGroups',
     45: 'AlterPartitionReassignments',
     46: 'ListPartitionReassignments',
+    48: 'DescribeClientQuotas',
 }

kafka/protocol/admin.py

@@ -1,7 +1,8 @@
 from __future__ import absolute_import
 
 from kafka.protocol.api import Request, Response
-from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String, CompactString, CompactArray, TaggedFields
+
+from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String, Float64, CompactString, CompactArray, TaggedFields
 
 
 class ApiVersionResponse_v0(Response):
@@ -1011,4 +1012,45 @@ class ListPartitionReassignmentsRequest_v0(Request):
 
 ListPartitionReassignmentsRequest = [ListPartitionReassignmentsRequest_v0]
 
-ListPartitionReassignmentsResponse = [ListPartitionReassignmentsResponse_v0]
+ListPartitionReassignmentsResponse = [ListPartitionReassignmentsResponse_v0]
+
+
+class DescribeClientQuotasResponse_v0(Request):
+    API_KEY = 48
+    API_VERSION = 0
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('error_code', Int16),
+        ('error_message', String('utf-8')),
+        ('entries', Array(
+            ('entity', Array(
+                ('entity_type', String('utf-8')),
+                ('entity_name', String('utf-8')))),
+            ('values', Array(
+                ('name', String('utf-8')),
+                ('value', Float64))))),
+    )
+
+
+class DescribeClientQuotasRequest_v0(Request):
+    API_KEY = 48
+    API_VERSION = 0
+    RESPONSE_TYPE = DescribeClientQuotasResponse_v0
+    SCHEMA = Schema(
+        ('components', Array(
+            ('entity_type', String('utf-8')),
+            ('match_type', Int8),
+            ('match', String('utf-8')),
+        )),
+        ('strict', Boolean)
+    )
+
+
+DescribeClientQuotasRequest = [
+    DescribeClientQuotasRequest_v0,
+]
+
+DescribeClientQuotasResponse = [
+    DescribeClientQuotasResponse_v0,
+]
+

kafka/protocol/types.py

@@ -77,6 +77,19 @@ def decode(cls, data):
         return _unpack(cls._unpack, data.read(8))
 
 
+class Float64(AbstractType):
+    _pack = struct.Struct('>d').pack
+    _unpack = struct.Struct('>d').unpack
+
+    @classmethod
+    def encode(cls, value):
+        return _pack(cls._pack, value)
+
+    @classmethod
+    def decode(cls, data):
+        return _unpack(cls._unpack, data.read(8))
+
+
 class String(AbstractType):
     def __init__(self, encoding='utf-8'):
         self.encoding = encoding
@@ -181,9 +194,10 @@ def __init__(self, *array_of):
     def encode(self, items):
         if items is None:
             return Int32.encode(-1)
+        encoded_items = [self.array_of.encode(item) for item in items]
         return b''.join(
-            [Int32.encode(len(items))] +
-            [self.array_of.encode(item) for item in items]
+            [Int32.encode(len(encoded_items))] +
+            encoded_items
         )
 
     def decode(self, data):