Protocol updates for 0.11.0.0 (#1127)

1 files changed, 84 insertions, 11 deletions

diff --git a/kafka/protocol/commit.py b/kafka/protocol/commit.py
index bcffe67..9d744c7 100644
--- a/kafka/protocol/commit.py
+++ b/kafka/protocol/commit.py
@@ -1,7 +1,7 @@
 from __future__ import absolute_import
 
 from .api import Request, Response
-from .types import Array, Int16, Int32, Int64, Schema, String
+from .types import Array, Int8, Int16, Int32, Int64, Schema, String
 
 
 class OffsetCommitResponse_v0(Response):
@@ -28,6 +28,19 @@ class OffsetCommitResponse_v2(Response):
     SCHEMA = OffsetCommitResponse_v1.SCHEMA
 
 
+class OffsetCommitResponse_v3(Response):
+    API_KEY = 8
+    API_VERSION = 3
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('topics', Array(
+            ('topic', String('utf-8')),
+            ('partitions', Array(
+                ('partition', Int32),
+                ('error_code', Int16)))))
+    )
+
+
 class OffsetCommitRequest_v0(Request):
     API_KEY = 8
     API_VERSION = 0  # Zookeeper-backed storage
@@ -81,10 +94,21 @@ class OffsetCommitRequest_v2(Request):
     DEFAULT_RETENTION_TIME = -1
 
 
-OffsetCommitRequest = [OffsetCommitRequest_v0, OffsetCommitRequest_v1,
-                       OffsetCommitRequest_v2]
-OffsetCommitResponse = [OffsetCommitResponse_v0, OffsetCommitResponse_v1,
-                        OffsetCommitResponse_v2]
+class OffsetCommitRequest_v3(Request):
+    API_KEY = 8
+    API_VERSION = 3
+    RESPONSE_TYPE = OffsetCommitResponse_v3
+    SCHEMA = OffsetCommitRequest_v2.SCHEMA
+
+
+OffsetCommitRequest = [
+    OffsetCommitRequest_v0, OffsetCommitRequest_v1,
+    OffsetCommitRequest_v2, OffsetCommitRequest_v3
+]
+OffsetCommitResponse = [
+    OffsetCommitResponse_v0, OffsetCommitResponse_v1,
+    OffsetCommitResponse_v2, OffsetCommitResponse_v3
+]
 
 
 class OffsetFetchResponse_v0(Response):
@@ -123,6 +147,22 @@ class OffsetFetchResponse_v2(Response):
     )
 
 
+class OffsetFetchResponse_v3(Response):
+    API_KEY = 9
+    API_VERSION = 3
+    SCHEMA = Schema(
+        ('throttle_time_ms', Int32),
+        ('topics', Array(
+            ('topic', String('utf-8')),
+            ('partitions', Array(
+                ('partition', Int32),
+                ('offset', Int64),
+                ('metadata', String('utf-8')),
+                ('error_code', Int16))))),
+        ('error_code', Int16)
+    )
+
+
 class OffsetFetchRequest_v0(Request):
     API_KEY = 9
     API_VERSION = 0  # zookeeper-backed storage
@@ -152,10 +192,21 @@ class OffsetFetchRequest_v2(Request):
     SCHEMA = OffsetFetchRequest_v1.SCHEMA
 
 
-OffsetFetchRequest = [OffsetFetchRequest_v0, OffsetFetchRequest_v1,
-    OffsetFetchRequest_v2]
-OffsetFetchResponse = [OffsetFetchResponse_v0, OffsetFetchResponse_v1,
-    OffsetFetchResponse_v2]
+class OffsetFetchRequest_v3(Request):
+    API_KEY = 9
+    API_VERSION = 3
+    RESPONSE_TYPE = OffsetFetchResponse_v3
+    SCHEMA = OffsetFetchRequest_v2.SCHEMA
+
+
+OffsetFetchRequest = [
+    OffsetFetchRequest_v0, OffsetFetchRequest_v1,
+    OffsetFetchRequest_v2, OffsetFetchRequest_v3,
+]
+OffsetFetchResponse = [
+    OffsetFetchResponse_v0, OffsetFetchResponse_v1,
+    OffsetFetchResponse_v2, OffsetFetchResponse_v3,
+]
 
 
 class GroupCoordinatorResponse_v0(Response):
@@ -169,6 +220,18 @@ class GroupCoordinatorResponse_v0(Response):
     )
 
 
+class GroupCoordinatorResponse_v1(Response):
+    API_KEY = 10
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('error_code', Int16),
+        ('error_message', String('utf-8')),
+        ('coordinator_id', Int32),
+        ('host', String('utf-8')),
+        ('port', Int32)
+    )
+
+
 class GroupCoordinatorRequest_v0(Request):
     API_KEY = 10
     API_VERSION = 0
@@ -178,5 +241,15 @@ class GroupCoordinatorRequest_v0(Request):
     )
 
 
-GroupCoordinatorRequest = [GroupCoordinatorRequest_v0]
-GroupCoordinatorResponse = [GroupCoordinatorResponse_v0]
+class GroupCoordinatorRequest_v1(Request):
+    API_KEY = 10
+    API_VERSION = 1
+    RESPONSE_TYPE = GroupCoordinatorResponse_v1
+    SCHEMA = Schema(
+        ('coordinator_key', String('utf-8')),
+        ('coordinator_type', Int8)
+    )
+
+
+GroupCoordinatorRequest = [GroupCoordinatorRequest_v0, GroupCoordinatorRequest_v1]
+GroupCoordinatorResponse = [GroupCoordinatorResponse_v0, GroupCoordinatorResponse_v1]