Merge pull request #762 from dpkp/metadata_v1

Use Metadata Request/Response v1 with 0.10+ brokers

1 files changed, 36 insertions, 4 deletions

diff --git a/kafka/protocol/metadata.py b/kafka/protocol/metadata.py
index 8063dda..2711abb 100644
--- a/kafka/protocol/metadata.py
+++ b/kafka/protocol/metadata.py
@@ -1,5 +1,5 @@
 from .struct import Struct
-from .types import Array, Int16, Int32, Schema, String
+from .types import Array, Boolean, Int16, Int32, Schema, String
 
 
 class MetadataResponse_v0(Struct):
@@ -22,14 +22,46 @@ class MetadataResponse_v0(Struct):
     )
 
 
+class MetadataResponse_v1(Struct):
+    API_KEY = 3
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('brokers', Array(
+            ('node_id', Int32),
+            ('host', String('utf-8')),
+            ('port', Int32),
+            ('rack', String('utf-8')))),
+        ('controller_id', Int32),
+        ('topics', Array(
+            ('error_code', Int16),
+            ('topic', String('utf-8')),
+            ('is_internal', Boolean),
+            ('partitions', Array(
+                ('error_code', Int16),
+                ('partition', Int32),
+                ('leader', Int32),
+                ('replicas', Array(Int32)),
+                ('isr', Array(Int32))))))
+    )
+
+
 class MetadataRequest_v0(Struct):
     API_KEY = 3
     API_VERSION = 0
     RESPONSE_TYPE = MetadataResponse_v0
     SCHEMA = Schema(
-        ('topics', Array(String('utf-8')))
+        ('topics', Array(String('utf-8'))) # Empty Array (len 0) for all topics
+    )
+
+
+class MetadataRequest_v1(Struct):
+    API_KEY = 3
+    API_VERSION = 1
+    RESPONSE_TYPE = MetadataResponse_v1
+    SCHEMA = Schema(
+        ('topics', Array(String('utf-8'))) # Null Array (len -1) for all topics
     )
 
 
-MetadataRequest = [MetadataRequest_v0]
-MetadataResponse = [MetadataResponse_v0]
+MetadataRequest = [MetadataRequest_v0, MetadataRequest_v1]
+MetadataResponse = [MetadataResponse_v0, MetadataResponse_v1]