Merge pull request dpkp#678 from dpkp/check_version_0_10

dpkp · dpkp · commit d401bde9ed64 · 2016-05-03T17:35:59.000-07:00
Add protocol support for ApiVersionRequest
diff --git a/kafka/conn.py b/kafka/conn.py
@@ -520,7 +520,7 @@ def check_version(self, timeout=2, strict=False):
         # vanilla MetadataRequest. If the server did not recognize the first
         # request, both will be failed with a ConnectionError that wraps
         # socket.error (32, 54, or 104)
-        from .protocol.admin import ListGroupsRequest
+        from .protocol.admin import ApiVersionRequest, ListGroupsRequest
         from .protocol.commit import OffsetFetchRequest, GroupCoordinatorRequest
         from .protocol.metadata import MetadataRequest
 
@@ -536,6 +536,7 @@ def filter(self, record):
         log.addFilter(log_filter)
 
         test_cases = [
+            ('0.10', ApiVersionRequest[0]()),
             ('0.9', ListGroupsRequest[0]()),
             ('0.8.2', GroupCoordinatorRequest[0]('kafka-python-default-group')),
             ('0.8.1', OffsetFetchRequest[0]('kafka-python-default-group', [])),
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
@@ -225,7 +225,7 @@ def __init__(self, *topics, **configs):
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':
             self.config['api_version'] = self._client.check_version()
-        assert self.config['api_version'] in ('0.9', '0.8.2', '0.8.1', '0.8.0'), 'Unrecognized api version'
+        assert self.config['api_version'] in ('0.10', '0.9', '0.8.2', '0.8.1', '0.8.0'), 'Unrecognized api version'
 
         # Convert api_version config to tuple for easy comparisons
         self.config['api_version'] = tuple(
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
@@ -268,7 +268,7 @@ def __init__(self, **configs):
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':
             self.config['api_version'] = client.check_version()
-        assert self.config['api_version'] in ('0.9', '0.8.2', '0.8.1', '0.8.0')
+        assert self.config['api_version'] in ('0.10', '0.9', '0.8.2', '0.8.1', '0.8.0')
 
         # Convert api_version config to tuple for easy comparisons
         self.config['api_version'] = tuple(
diff --git a/kafka/protocol/admin.py b/kafka/protocol/admin.py
@@ -2,6 +2,28 @@
 from .types import Array, Bytes, Int16, Schema, String
 
 
+class ApiVersionResponse_v0(Struct):
+    API_KEY = 18
+    API_VERSION = 0
+    SCHEMA = Schema(
+        ('error_code', Int16),
+        ('api_versions', Array(
+            ('api_key', Int16),
+            ('min_version', Int16),
+            ('max_version', Int16))))
+
+
+class ApiVersionRequest_v0(Struct):
+    API_KEY = 18
+    API_VERSION = 0
+    RESPONSE_TYPE = ApiVersionResponse_v0
+    SCHEMA = Schema()
+
+
+ApiVersionRequest = [ApiVersionRequest_v0]
+ApiVersionResponse = [ApiVersionResponse_v0]
+
+
 class ListGroupsResponse_v0(Struct):
     API_KEY = 16
     API_VERSION = 0
