CreateTopicsRequest / Response v1create_topics_v1

1 files changed, 34 insertions, 3 deletions

diff --git a/kafka/protocol/admin.py b/kafka/protocol/admin.py
index 99ec177..89ea739 100644
--- a/kafka/protocol/admin.py
+++ b/kafka/protocol/admin.py
@@ -1,7 +1,7 @@
 from __future__ import absolute_import
 
 from .struct import Struct
-from .types import Array, Bytes, Int16, Int32, Schema, String
+from .types import Array, Boolean, Bytes, Int16, Int32, Schema, String
 
 
 class ApiVersionResponse_v0(Struct):
@@ -37,6 +37,17 @@ class CreateTopicsResponse_v0(Struct):
     )
 
 
+class CreateTopicsResponse_v1(Struct):
+    API_KEY = 19
+    API_VERSION = 1
+    SCHEMA = Schema(
+        ('topic_error_codes', Array(
+            ('topic', String('utf-8')),
+            ('error_code', Int16),
+            ('error_message', String('utf-8'))))
+    )
+
+
 class CreateTopicsRequest_v0(Struct):
     API_KEY = 19
     API_VERSION = 0
@@ -56,8 +67,28 @@ class CreateTopicsRequest_v0(Struct):
     )
 
 
-CreateTopicsRequest = [CreateTopicsRequest_v0]
-CreateTopicsResponse = [CreateTopicsResponse_v0]
+class CreateTopicsRequest_v1(Struct):
+    API_KEY = 19
+    API_VERSION = 1
+    RESPONSE_TYPE = CreateTopicsResponse_v1
+    SCHEMA = Schema(
+        ('create_topic_requests', Array(
+            ('topic', String('utf-8')),
+            ('num_partitions', Int32),
+            ('replication_factor', Int16),
+            ('replica_assignment', Array(
+                ('partition_id', Int32),
+                ('replicas', Array(Int32)))),
+            ('configs', Array(
+                ('config_key', String('utf-8')),
+                ('config_value', String('utf-8')))))),
+        ('timeout', Int32),
+        ('validate_only', Boolean)
+    )
+
+
+CreateTopicsRequest = [CreateTopicsRequest_v0, CreateTopicsRequest_v1]
+CreateTopicsResponse = [CreateTopicsResponse_v0, CreateTopicsRequest_v1]
 
 
 class DeleteTopicsResponse_v0(Struct):