Repository: Parsely/pykafka (python)

Issue:
Support updated versions of MetadataRequest
pykafka currently only supports `MetadataRequest` and `MetadataResponse` v0. It needs to handle requests from later versions of the protocol. #750 is blocked by this.

Pull request description:
Metadata request version support
This pull request uses `ApiVersionAware` to support all currently existing versions of the `MetadataRequest` and `MetadataResponse`. Among other convenient features, this allows pykafka to (when supported) discover the ID of the controller broker, which is necessary for certain administrative API requests.

Fixes #755 

Patch:
diff --git a/pykafka/broker.py b/pykafka/broker.py
index a2ad55d..0ee8f41 100644
--- a/pykafka/broker.py
+++ b/pykafka/broker.py
@@ -355,6 +355,9 @@ class Broker(object):
         :param topics: The topic names for which to request metadata
         :type topics: Iterable of `bytes`
         """
+        request_class = MetadataRequest.get_version_impl(self._api_versions)
+        response_class = MetadataResponse.get_version_impl(self._api_versions)
+
         max_retries = 3
         for i in range(max_retries):
             if i > 0:
@@ -362,8 +365,8 @@ class Broker(object):
             time.sleep(i)
 
             try:
-                future = self._req_handler.request(MetadataRequest(topics=topics))
-                response = future.get(MetadataResponse)
+                future = self._req_handler.request(request_class(topics=topics))
+                response = future.get(response_class)
             except SocketDisconnectedError:
                 log.warning("Encountered SocketDisconnectedError while requesting "
                             "metadata from broker %s:%s. Continuing.",
diff --git a/pykafka/cluster.py b/pykafka/cluster.py
index cac2123..ff8972a 100644
--- a/pykafka/cluster.py
+++ b/pykafka/cluster.py
@@ -205,6 +205,7 @@ class Cluster(object):
         self._max_connection_retries_offset_mgr = 8
         self._broker_version = broker_version
         self._api_versions = None
+        self.controller_broker = None
         if ':' in self._source_address:
             self._source_port = int(self._source_address.split(':')[1])
         self.fetch_api_versions()
@@ -357,12 +358,14 @@ class Cluster(object):
                 log.exception(e)
                 return []
 
-    def _update_brokers(self, broker_metadata):
+    def _update_brokers(self, broker_metadata, controller_id):
         """Update brokers with fresh metadata.
 
         :param broker_metadata: Metadata for all brokers.
         :type broker_metadata: Dict of `{name: metadata}` where `metadata` is
             :class:`pykafka.protocol.BrokerMetadata` and `name` is str.
+        :param controller_id: The ID of the cluster's controller broker, if applicable
+        :type controller_id: int
         """
         # FIXME: A cluster with no topics returns no brokers in metadata
         # Remove old brokers
@@ -404,6 +407,11 @@ class Cluster(object):
                 #       Figure out and implement update/disconnect/reconnect if
                 #       needed.
                 raise Exception('Broker host/port change detected! %s', broker)
+        if controller_id is not None:
+            if controller_id not in self._brokers:
+                raise KeyError("Controller ID {} not present in cluster".format(
+                    controller_id))
+            self.controller_broker = self._brokers[controller_id]
 
     def get_managed_group_descriptions(self):
         """Return detailed descriptions of all managed consumer groups on this cluster
@@ -495,7 +503,7 @@ class Cluster(object):
                             'a topic in the cluster. See '
                             'https://issues.apache.org/jira/browse/KAFKA-2154 '
                             'for information.')
-            self._update_brokers(metadata.brokers)
+            self._update_brokers(metadata.brokers, metadata.controller_id)
             try:
                 self._topics._update_topics(metadata.topics)
             except LeaderNotFoundError:
diff --git a/pykafka/protocol.py b/pykafka/protocol.py
index 54653e2..ceabac6 100644
--- a/pykafka/protocol.py
+++ b/pykafka/protocol.py
@@ -432,9 +432,15 @@ class MetadataRequest(Request):
         MetadataRequest => [TopicName]
             TopicName => string
     """
+    API_VERSION = 0
     API_KEY = 3
 
-    def __init__(self, topics=None):
+    @classmethod
+    def get_versions(cls):
+        return {0: MetadataRequest, 1: MetadataRequestV1, 2: MetadataRequestV2,
+                3: MetadataRequestV3, 4: MetadataRequestV4, 5: MetadataRequestV5}
+
+    def __init__(self, topics=None, *kwargs):
         """Create a new MetadataRequest
 
         :param topics: Topics to query. Leave empty for all available topics.
@@ -445,27 +451,91 @@ class MetadataRequest(Request):
         """Length of the serialized message, in bytes"""
         return self.HEADER_LEN + 4 + sum(len(t) + 2 for t in self.topics)
 
-    def get_bytes(self):
-        """Serialize the message
+    def _topics_len(self):
+        return len(self.topics)
 
-        :returns: Serialized message
-        :rtype: :class:`bytearray`
-        """
+    def _serialize(self):
         output = bytearray(len(self))
-        self._write_header(output)
-        struct.pack_into('!i', output, self.HEADER_LEN, len(self.topics))
+        self._write_header(output, api_version=self.API_VERSION)
+        struct.pack_into('!i', output, self.HEADER_LEN, self._topics_len())
         offset = self.HEADER_LEN + 4
         for t in self.topics:
             tlen = len(t)
             struct.pack_into('!h%ds' % tlen, output, offset, tlen, t)
             offset += 2 + tlen
+        return output, offset
+
+    def get_bytes(self):
+        """Serialize the message
+
+        :returns: Serialized message
+        :rtype: :class:`bytearray`
+        """
+        output, _ = self._serialize()
         return output
 
 
+class MetadataRequestV1(MetadataRequest):
+    API_VERSION = 1
+
+    def _topics_len(self):
+        # v1 and higher require a null array, not an empty array, to select all topics
+        return len(self.topics) or -1
+
+
+class MetadataRequestV2(MetadataRequestV1):
+    API_VERSION = 2
+
+
+class MetadataRequestV3(MetadataRequestV2):
+    API_VERSION = 3
+
+
+class MetadataRequestV4(MetadataRequestV3):
+    """Metadata Request
+
+    Specification::
+
+    Metadata Request (Version: 4) => [topics] allow_auto_topic_creation
+        topics => STRING
+        allow_auto_topic_creation => BOOLEAN
+    """
+    API_VERSION = 4
+
+    def __init__(self, topics=None, allow_topic_autocreation=True):
+        """Create a new MetadataRequest
+
+        :param topics: Topics to query. Leave empty for all available topics.
+        :param allow_topic_autocreation: If this and the broker config
+            'auto.create.topics.enable' are true, topics that don't exist will be created
+            by the broker. Otherwise, no topics will be created by the broker.
+        """
+        super(MetadataRequestV4, self).__init__(topics=topics)
+        self.allow_topic_autocreation = allow_topic_autocreation
+
+    def __len__(self):
+        return super(MetadataRequestV4, self).__len__() + 1
+
+    def get_bytes(self):
+        output, offset = self._serialize()
+        struct.pack_into('!b', output, offset, self.allow_topic_autocreation)
+        return output
+
+
+class MetadataRequestV5(MetadataRequestV4):
+    API_VERSION = 5
+
+
 BrokerMetadata = namedtuple('BrokerMetadata', ['id', 'host', 'port'])
+BrokerMetadataV1 = namedtuple('BrokerMetadataV1', ['id', 'host', 'port', 'rack'])
 TopicMetadata = namedtuple('TopicMetadata', ['name', 'partitions', 'err'])
+TopicMetadataV1 = namedtuple('TopicMetadataV1', ['name', 'is_internal', 'partitions',
+                                                 'err'])
 PartitionMetadata = namedtuple('PartitionMetadata',
                                ['id', 'leader', 'replicas', 'isr', 'err'])
+PartitionMetadataV5 = namedtuple('PartitionMetadataV5',
+                                 ['id', 'leader', 'replicas', 'isr', 'offline_replicas',
+                                  'err'])
 
 
 class MetadataResponse(Response):
@@ -473,20 +543,28 @@ class MetadataResponse(Response):
 
     Specification::
 
-        MetadataResponse => [Broker][TopicMetadata]
-          Broker => NodeId Host Port
-          NodeId => int32
-          Host => string
-          Port => int32
-          TopicMetadata => TopicErrorCode TopicName [PartitionMetadata]
-          TopicErrorCode => int16
-          PartitionMetadata => PartitionErrorCode PartitionId Leader Replicas Isr
-          PartitionErrorCode => int16
-          PartitionId => int32
-          Leader => int32
-          Replicas => [int32]
-          Isr => [int32]
+    Metadata Response (Version: 0) => [brokers] [topic_metadata]
+        brokers => node_id host port
+            node_id => INT32
+            host => STRING
+            port => INT32
+        topic_metadata => error_code topic [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            partition_metadata => error_code partition leader [replicas] [isr]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
     """
+    API_KEY = 3
+
+    @classmethod
+    def get_versions(cls):
+        return {0: MetadataResponse, 1: MetadataResponseV1, 2: MetadataResponseV2,
+                3: MetadataResponseV3, 4: MetadataResponseV4, 5: MetadataResponseV5}
+
     def __init__(self, buff):
         """Deserialize into a new Response
 
@@ -496,18 +574,215 @@ class MetadataResponse(Response):
         fmt = '[iSi] [hS [hii [i] [i] ] ]'
         response = struct_helpers.unpack_from(fmt, buff, 0)
         broker_info, topics = response
-
+        self._populate(broker_info, topics)
+
+    def _populate(self,
+                  broker_info,
+                  topics,
+                  controller_id=None,
+                  cluster_id=None,
+                  throttle_time_ms=0):
+        self.throttle_time_ms = throttle_time_ms
+        self._build_broker_metas(broker_info)
+        self.cluster_id = cluster_id
+        self.controller_id = controller_id
+        self._build_topic_metas(topics)
+
+    def _build_topic_metas(self, topics):
+        self.topics = {}
+        for (err, name, partitions) in topics:
+            self.topics[name] = TopicMetadata(name,
+                                              self._build_partition_metas(partitions),
+                                              err)
+
+    def _build_partition_metas(self, partitions):
+        part_metas = {}
+        for (p_err, id_, leader, replicas, isr) in partitions:
+            part_metas[id_] = PartitionMetadata(id_, leader, replicas,
+                                                isr, p_err)
+        return part_metas
+
+    def _build_broker_metas(self, broker_info):
         self.brokers = {}
         for (id_, host, port) in broker_info:
             self.brokers[id_] = BrokerMetadata(id_, host, port)
 
+
+class MetadataResponseV1(MetadataResponse):
+    """Response from MetadataRequest
+
+    Specification::
+
+    Metadata Response (Version: 1) => [brokers] controller_id [topic_metadata]
+        brokers => node_id host port rack
+            node_id => INT32
+            host => STRING
+            port => INT32
+            rack => NULLABLE_STRING  (new since v0)
+        controller_id => INT32  (new since v0)
+        topic_metadata => error_code topic is_internal [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            is_internal => BOOLEAN  (new since v0)
+            partition_metadata => error_code partition leader [replicas] [isr]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
+    """
+    def __init__(self, buff):
+        fmt = '[iSiS] i [hSb [hii [i] [i] ] ]'
+        response = struct_helpers.unpack_from(fmt, buff, 0)
+        broker_info, controller_id, topics = response
+        self._populate(broker_info, topics, controller_id=controller_id)
+
+    def _build_topic_metas(self, topics):
         self.topics = {}
-        for (err, name, partitions) in topics:
-            part_metas = {}
-            for (p_err, id_, leader, replicas, isr) in partitions:
-                part_metas[id_] = PartitionMetadata(id_, leader, replicas,
-                                                    isr, p_err)
-            self.topics[name] = TopicMetadata(name, part_metas, err)
+        for (err, name, is_internal, partitions) in topics:
+            self.topics[name] = TopicMetadataV1(name,
+                                                is_internal,
+                                                self._build_partition_metas(partitions),
+                                                err)
+
+    def _build_broker_metas(self, broker_info):
+        self.brokers = {}
+        for (id_, host, port, rack) in broker_info:
+            self.brokers[id_] = BrokerMetadataV1(id_, host, port, rack)
+
+
+class MetadataResponseV2(MetadataResponseV1):
+    """Response from MetadataRequest
+
+    Specification::
+
+    Metadata Response (Version: 2) => [brokers] cluster_id controller_id [topic_metadata]
+        brokers => node_id host port rack
+            node_id => INT32
+            host => STRING
+            port => INT32
+            rack => NULLABLE_STRING
+        cluster_id => NULLABLE_STRING  (new since v1)
+        controller_id => INT32
+        topic_metadata => error_code topic is_internal [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            is_internal => BOOLEAN
+            partition_metadata => error_code partition leader [replicas] [isr]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
+    """
+    def __init__(self, buff):
+        fmt = '[iSiS] Si [hSb [hii [i] [i] ] ]'
+        response = struct_helpers.unpack_from(fmt, buff, 0)
+        broker_info, cluster_id, controller_id, topics = response
+        self._populate(broker_info, topics, controller_id=controller_id,
+                       cluster_id=cluster_id)
+
+
+class MetadataResponseV3(MetadataResponseV2):
+    """Response from MetadataRequest
+
+    Specification::
+
+    Metadata Response (Version: 3) => throttle_time_ms [brokers] cluster_id controller_id [topic_metadata]
+        throttle_time_ms => INT32  (new since v2)
+        brokers => node_id host port rack
+            node_id => INT32
+            host => STRING
+            port => INT32
+            rack => NULLABLE_STRING
+        cluster_id => NULLABLE_STRING
+        controller_id => INT32
+        topic_metadata => error_code topic is_internal [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            is_internal => BOOLEAN
+            partition_metadata => error_code partition leader [replicas] [isr]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
+    """
+    def __init__(self, buff):
+        fmt = 'i [iSiS] Si [hSb [hii [i] [i] ] ]'
+        response = struct_helpers.unpack_from(fmt, buff, 0)
+        throttle_time_ms, broker_info, cluster_id, controller_id, topics = response
+        self._populate(broker_info, topics, controller_id=controller_id,
+                       cluster_id=cluster_id, throttle_time_ms=throttle_time_ms)
+
+
+class MetadataResponseV4(MetadataResponseV3):
+    """Response from MetadataRequest
+
+    Specification::
+
+    Metadata Response (Version: 4) => throttle_time_ms [brokers] cluster_id controller_id [topic_metadata]
+        throttle_time_ms => INT32
+        brokers => node_id host port rack
+            node_id => INT32
+            host => STRING
+            port => INT32
+            rack => NULLABLE_STRING
+        cluster_id => NULLABLE_STRING
+        controller_id => INT32
+        topic_metadata => error_code topic is_internal [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            is_internal => BOOLEAN
+            partition_metadata => error_code partition leader [replicas] [isr]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
+    """
+    pass
+
+
+class MetadataResponseV5(MetadataResponseV4):
+    """Response from MetadataRequest
+
+    Specification::
+
+    Metadata Response (Version: 5) => throttle_time_ms [brokers] cluster_id controller_id [topic_metadata]
+        throttle_time_ms => INT32
+        brokers => node_id host port rack
+            node_id => INT32
+            host => STRING
+            port => INT32
+            rack => NULLABLE_STRING
+        cluster_id => NULLABLE_STRING
+        controller_id => INT32
+        topic_metadata => error_code topic is_internal [partition_metadata]
+            error_code => INT16
+            topic => STRING
+            is_internal => BOOLEAN
+            partition_metadata => error_code partition leader [replicas] [isr] [offline_replicas]
+                error_code => INT16
+                partition => INT32
+                leader => INT32
+                replicas => INT32
+                isr => INT32
+                offline_replicas => INT32  (new since v4)
+    """
+    def __init__(self, buff):
+        fmt = 'i [iSiS] Si [hSb [hii [i] [i] [i]] ]'
+        response = struct_helpers.unpack_from(fmt, buff, 0)
+        throttle_time_ms, broker_info, cluster_id, controller_id, topics = response
+        self._populate(broker_info, topics, controller_id=controller_id,
+                       cluster_id=cluster_id, throttle_time_ms=throttle_time_ms)
+
+    def _build_partition_metas(self, partitions):
+        part_metas = {}
+        for (p_err, id_, leader, replicas, isr, offline_replicas) in partitions:
+            part_metas[id_] = PartitionMetadataV5(id_, leader, replicas,
+                                                  isr, offline_replicas, p_err)
+        return part_metas
 
 
 ##


Test patch:
diff --git a/tests/pykafka/test_protocol.py b/tests/pykafka/test_protocol.py
index 4dbe582..dd1daa4 100644
--- a/tests/pykafka/test_protocol.py
+++ b/tests/pykafka/test_protocol.py
@@ -132,6 +132,367 @@ class TestMetadataAPI(unittest2.TestCase):
         self.assertEqual(response.topics[b'test'].err, 3)
 
 
+class TestMetadataAPIV1(unittest2.TestCase):
+    maxDiff = None
+
+    def test_request(self):
+        req = protocol.MetadataRequestV1()
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                # header
+                b'\x00\x00\x00\x15'  # len(buffer)
+                b'\x00\x03'  # ApiKey
+                b'\x00\x01'  # api version
+                b'\x00\x00\x00\x00'  # correlation id
+                b'\x00\x07'  # len(client id)
+                    b'pykafka'  # client id  # noqa
+                # end header
+
+                b'\xff\xff\xff\xff'  # len(topics)
+            )
+        )
+
+    def test_response(self):
+        cluster = protocol.MetadataResponseV1(
+            buffer(
+                b'\x00\x00\x00\x01'  # len(brokers)
+                    b'\x00\x00\x00\x00'  # node id # noqa
+                    b'\x00\x09'  # len(host)
+                        b'localhost'  # host
+                    b'\x00\x00#\x84'  # port
+                    b'\xff\xff'  # len(rack)
+                b'\x00\x00\x00\x00'  # controller_id
+                b'\x00\x00\x00\x01'  # len(topic metadata)
+                    b'\x00\x00'  # error code
+                    b'\x00\x04'  # len(topic name)
+                        b'test'  # topic name
+                    b'\x00'  # is_internal
+                    b'\x00\x00\x00\x02'  # len(partition metadata)
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x00'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replica
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x01'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replicas
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+            )
+        )
+        self.assertEqual(cluster.brokers[0].host, b'localhost')
+        self.assertEqual(cluster.brokers[0].port, 9092)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].leader,
+                         cluster.brokers[0].id)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].isr,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.brokers[0].rack, None)
+        self.assertEqual(cluster.controller_id, 0)
+        self.assertEqual(cluster.topics[b'test'].is_internal, False)
+
+
+class TestMetadataAPIV2(unittest2.TestCase):
+    maxDiff = None
+
+    def test_request(self):
+        req = protocol.MetadataRequestV2()
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                # header
+                b'\x00\x00\x00\x15'  # len(buffer)
+                b'\x00\x03'  # ApiKey
+                b'\x00\x02'  # api version
+                b'\x00\x00\x00\x00'  # correlation id
+                b'\x00\x07'  # len(client id)
+                    b'pykafka'  # client id  # noqa
+                # end header
+
+                b'\xff\xff\xff\xff'  # len(topics)
+            )
+        )
+
+    def test_response(self):
+        cluster = protocol.MetadataResponseV2(
+            buffer(
+                b'\x00\x00\x00\x01'  # len(brokers)
+                    b'\x00\x00\x00\x00'  # node id # noqa
+                    b'\x00\x09'  # len(host)
+                        b'localhost'  # host
+                    b'\x00\x00#\x84'  # port
+                    b'\xff\xff'  # len(rack)
+                b'\x00\x01'  # len(cluster_id)
+                    b'a'  # cluster_id
+                b'\x00\x00\x00\x00'  # controller_id
+                b'\x00\x00\x00\x01'  # len(topic metadata)
+                    b'\x00\x00'  # error code
+                    b'\x00\x04'  # len(topic name)
+                        b'test'  # topic name
+                    b'\x00'  # is_internal
+                    b'\x00\x00\x00\x02'  # len(partition metadata)
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x00'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replica
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x01'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replicas
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+            )
+        )
+        self.assertEqual(cluster.brokers[0].host, b'localhost')
+        self.assertEqual(cluster.brokers[0].port, 9092)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].leader,
+                         cluster.brokers[0].id)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].isr,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.brokers[0].rack, None)
+        self.assertEqual(cluster.controller_id, 0)
+        self.assertEqual(cluster.cluster_id, b"a")
+        self.assertEqual(cluster.topics[b'test'].is_internal, False)
+
+
+class TestMetadataAPIV3(unittest2.TestCase):
+    maxDiff = None
+
+    def test_request(self):
+        req = protocol.MetadataRequestV3()
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                # header
+                b'\x00\x00\x00\x15'  # len(buffer)
+                b'\x00\x03'  # ApiKey
+                b'\x00\x03'  # api version
+                b'\x00\x00\x00\x00'  # correlation id
+                b'\x00\x07'  # len(client id)
+                    b'pykafka'  # client id  # noqa
+                # end header
+
+                b'\xff\xff\xff\xff'  # len(topics)
+            )
+        )
+
+    def test_response(self):
+        cluster = protocol.MetadataResponseV3(
+            buffer(
+                b'\x00\x00\x00\x00'  # throttle_time_ms
+                b'\x00\x00\x00\x01'  # len(brokers)
+                    b'\x00\x00\x00\x00'  # node id # noqa
+                    b'\x00\x09'  # len(host)
+                        b'localhost'  # host
+                    b'\x00\x00#\x84'  # port
+                    b'\xff\xff'  # len(rack)
+                b'\x00\x01'  # len(cluster_id)
+                    b'a'  # cluster_id
+                b'\x00\x00\x00\x00'  # controller_id
+                b'\x00\x00\x00\x01'  # len(topic metadata)
+                    b'\x00\x00'  # error code
+                    b'\x00\x04'  # len(topic name)
+                        b'test'  # topic name
+                    b'\x00'  # is_internal
+                    b'\x00\x00\x00\x02'  # len(partition metadata)
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x00'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replica
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x01'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replicas
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+            )
+        )
+        self.assertEqual(cluster.brokers[0].host, b'localhost')
+        self.assertEqual(cluster.brokers[0].port, 9092)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].leader,
+                         cluster.brokers[0].id)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].isr,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.brokers[0].rack, None)
+        self.assertEqual(cluster.throttle_time_ms, 0)
+        self.assertEqual(cluster.controller_id, 0)
+        self.assertEqual(cluster.cluster_id, b"a")
+        self.assertEqual(cluster.topics[b'test'].is_internal, False)
+
+
+class TestMetadataAPIV4(unittest2.TestCase):
+    maxDiff = None
+
+    def test_request(self):
+        req = protocol.MetadataRequestV4()
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                # header
+                b'\x00\x00\x00\x16'  # len(buffer)
+                b'\x00\x03'  # ApiKey
+                b'\x00\x04'  # api version
+                b'\x00\x00\x00\x00'  # correlation id
+                b'\x00\x07'  # len(client id)
+                    b'pykafka'  # client id  # noqa
+                # end header
+
+                b'\xff\xff\xff\xff'  # len(topics)
+                b'\x01'  # allow_topic_autocreation
+            )
+        )
+
+    def test_response(self):
+        cluster = protocol.MetadataResponseV4(
+            buffer(
+                b'\x00\x00\x00\x00'  # throttle_time_ms
+                b'\x00\x00\x00\x01'  # len(brokers)
+                    b'\x00\x00\x00\x00'  # node id # noqa
+                    b'\x00\x09'  # len(host)
+                        b'localhost'  # host
+                    b'\x00\x00#\x84'  # port
+                    b'\xff\xff'  # len(rack)
+                b'\x00\x01'  # len(cluster_id)
+                    b'a'  # cluster_id
+                b'\x00\x00\x00\x00'  # controller_id
+                b'\x00\x00\x00\x01'  # len(topic metadata)
+                    b'\x00\x00'  # error code
+                    b'\x00\x04'  # len(topic name)
+                        b'test'  # topic name
+                    b'\x00'  # is_internal
+                    b'\x00\x00\x00\x02'  # len(partition metadata)
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x00'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replica
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x01'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replicas
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+            )
+        )
+        self.assertEqual(cluster.brokers[0].host, b'localhost')
+        self.assertEqual(cluster.brokers[0].port, 9092)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].leader,
+                         cluster.brokers[0].id)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].isr,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.brokers[0].rack, None)
+        self.assertEqual(cluster.throttle_time_ms, 0)
+        self.assertEqual(cluster.controller_id, 0)
+        self.assertEqual(cluster.cluster_id, b"a")
+        self.assertEqual(cluster.topics[b'test'].is_internal, False)
+
+
+class TestMetadataAPIV5(unittest2.TestCase):
+    maxDiff = None
+
+    def test_request(self):
+        req = protocol.MetadataRequestV5()
+        msg = req.get_bytes()
+        self.assertEqual(
+            msg,
+            bytearray(
+                # header
+                b'\x00\x00\x00\x16'  # len(buffer)
+                b'\x00\x03'  # ApiKey
+                b'\x00\x05'  # api version
+                b'\x00\x00\x00\x00'  # correlation id
+                b'\x00\x07'  # len(client id)
+                    b'pykafka'  # client id  # noqa
+                # end header
+
+                b'\xff\xff\xff\xff'  # len(topics)
+                b'\x01'  # allow_topic_autocreation
+            )
+        )
+
+    def test_response(self):
+        cluster = protocol.MetadataResponseV5(
+            buffer(
+                b'\x00\x00\x00\x00'  # throttle_time_ms
+                b'\x00\x00\x00\x01'  # len(brokers)
+                    b'\x00\x00\x00\x00'  # node id # noqa
+                    b'\x00\x09'  # len(host)
+                        b'localhost'  # host
+                    b'\x00\x00#\x84'  # port
+                    b'\xff\xff'  # len(rack)
+                b'\x00\x01'  # len(cluster_id)
+                    b'a'  # cluster_id
+                b'\x00\x00\x00\x00'  # controller_id
+                b'\x00\x00\x00\x01'  # len(topic metadata)
+                    b'\x00\x00'  # error code
+                    b'\x00\x04'  # len(topic name)
+                        b'test'  # topic name
+                    b'\x00'  # is_internal
+                    b'\x00\x00\x00\x02'  # len(partition metadata)
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x00'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replica
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                            b'\x00\x00\x00\x01'  # len(offline_replicas)
+                                b'\x00\x00\x00\x00'  # offline_replicas
+                        b'\x00\x00'  # partition error code
+                            b'\x00\x00\x00\x01'  # partition id
+                            b'\x00\x00\x00\x00'  # leader
+                            b'\x00\x00\x00\x01'  # len(replicas)
+                                b'\x00\x00\x00\x00'  # replicas
+                            b'\x00\x00\x00\x01'  # len(isr)
+                                b'\x00\x00\x00\x00'  # isr
+                            b'\x00\x00\x00\x01'  # len(offline_replicas)
+                                b'\x00\x00\x00\x00'  # offline_replicas
+            )
+        )
+        self.assertEqual(cluster.brokers[0].host, b'localhost')
+        self.assertEqual(cluster.brokers[0].port, 9092)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].leader,
+                         cluster.brokers[0].id)
+        self.assertEqual(cluster.topics[b'test'].partitions[0].replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].isr,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.topics[b'test'].partitions[0].offline_replicas,
+                         [cluster.brokers[0].id])
+        self.assertEqual(cluster.brokers[0].rack, None)
+        self.assertEqual(cluster.throttle_time_ms, 0)
+        self.assertEqual(cluster.controller_id, 0)
+        self.assertEqual(cluster.cluster_id, b"a")
+        self.assertEqual(cluster.topics[b'test'].is_internal, False)
+
+
 class TestProduceAPI(unittest2.TestCase):
     maxDiff = None