Derive all api classes from Request / Response base classes (#1030)

Author: dpkp

kafka/client.py

@@ -257,18 +257,14 @@ def failed_payloads(payloads):
                 continue
 
             request = encoder_fn(payloads=broker_payloads)
-            # decoder_fn=None signal that the server is expected to not
-            # send a response.  This probably only applies to
-            # ProduceRequest w/ acks = 0
-            expect_response = (decoder_fn is not None)
-            future = conn.send(request, expect_response=expect_response)
+            future = conn.send(request)
 
             if future.failed():
                 refresh_metadata = True
                 failed_payloads(broker_payloads)
                 continue
 
-            if not expect_response:
+            if not request.expect_response():
                 for payload in broker_payloads:
                     topic_partition = (str(payload.topic), payload.partition)
                     responses[topic_partition] = None

kafka/client_async.py

@@ -464,12 +464,7 @@ def send(self, node_id, request):
         if not self._maybe_connect(node_id):
             return Future().failure(Errors.NodeNotReadyError(node_id))
 
-        # Every request gets a response, except one special case:
-        expect_response = True
-        if isinstance(request, tuple(ProduceRequest)) and request.required_acks == 0:
-            expect_response = False
-
-        return self._conns[node_id].send(request, expect_response=expect_response)
+        return self._conns[node_id].send(request)
 
     def poll(self, timeout_ms=None, future=None, sleep=True, delayed_tasks=True):
         """Try to read and write to sockets.

kafka/conn.py

@@ -525,7 +525,7 @@ def close(self, error=None):
             ifr.future.failure(error)
         self.config['state_change_callback'](self)
 
-    def send(self, request, expect_response=True):
+    def send(self, request):
         """send request, return Future()
 
         Can block on network if request is larger than send_buffer_bytes
@@ -537,9 +537,9 @@ def send(self, request, expect_response=True):
             return future.failure(Errors.ConnectionError(str(self)))
         elif not self.can_send_more():
             return future.failure(Errors.TooManyInFlightRequests(str(self)))
-        return self._send(request, expect_response=expect_response)
+        return self._send(request)
 
-    def _send(self, request, expect_response=True):
+    def _send(self, request):
         assert self.state in (ConnectionStates.AUTHENTICATING, ConnectionStates.CONNECTED)
         future = Future()
         correlation_id = self._next_correlation_id()
@@ -569,7 +569,7 @@ def _send(self, request, expect_response=True):
             return future.failure(error)
         log.debug('%s Request %d: %s', self, correlation_id, request)
 
-        if expect_response:
+        if request.expect_response():
             ifr = InFlightRequest(request=request,
                                   correlation_id=correlation_id,
                                   response_type=request.RESPONSE_TYPE,

kafka/protocol/admin.py

@@ -1,10 +1,10 @@
 from __future__ import absolute_import
 
-from .struct import Struct
+from .api import Request, Response
 from .types import Array, Boolean, Bytes, Int16, Int32, Schema, String
 
 
-class ApiVersionResponse_v0(Struct):
+class ApiVersionResponse_v0(Response):
     API_KEY = 18
     API_VERSION = 0
     SCHEMA = Schema(
@@ -16,7 +16,7 @@ class ApiVersionResponse_v0(Struct):
     )
 
 
-class ApiVersionRequest_v0(Struct):
+class ApiVersionRequest_v0(Request):
     API_KEY = 18
     API_VERSION = 0
     RESPONSE_TYPE = ApiVersionResponse_v0
@@ -27,7 +27,7 @@ class ApiVersionRequest_v0(Struct):
 ApiVersionResponse = [ApiVersionResponse_v0]
 
 
-class CreateTopicsResponse_v0(Struct):
+class CreateTopicsResponse_v0(Response):
     API_KEY = 19
     API_VERSION = 0
     SCHEMA = Schema(
@@ -37,7 +37,7 @@ class CreateTopicsResponse_v0(Struct):
     )
 
 
-class CreateTopicsResponse_v1(Struct):
+class CreateTopicsResponse_v1(Response):
     API_KEY = 19
     API_VERSION = 1
     SCHEMA = Schema(
@@ -48,7 +48,7 @@ class CreateTopicsResponse_v1(Struct):
     )
 
 
-class CreateTopicsRequest_v0(Struct):
+class CreateTopicsRequest_v0(Request):
     API_KEY = 19
     API_VERSION = 0
     RESPONSE_TYPE = CreateTopicsResponse_v0
@@ -67,7 +67,7 @@ class CreateTopicsRequest_v0(Struct):
     )
 
 
-class CreateTopicsRequest_v1(Struct):
+class CreateTopicsRequest_v1(Request):
     API_KEY = 19
     API_VERSION = 1
     RESPONSE_TYPE = CreateTopicsResponse_v1
@@ -91,7 +91,7 @@ class CreateTopicsRequest_v1(Struct):
 CreateTopicsResponse = [CreateTopicsResponse_v0, CreateTopicsRequest_v1]
 
 
-class DeleteTopicsResponse_v0(Struct):
+class DeleteTopicsResponse_v0(Response):
     API_KEY = 20
     API_VERSION = 0
     SCHEMA = Schema(
@@ -101,7 +101,7 @@ class DeleteTopicsResponse_v0(Struct):
     )
 
 
-class DeleteTopicsRequest_v0(Struct):
+class DeleteTopicsRequest_v0(Request):
     API_KEY = 20
     API_VERSION = 0
     RESPONSE_TYPE = DeleteTopicsResponse_v0
@@ -115,7 +115,7 @@ class DeleteTopicsRequest_v0(Struct):
 DeleteTopicsResponse = [DeleteTopicsResponse_v0]
 
 
-class ListGroupsResponse_v0(Struct):
+class ListGroupsResponse_v0(Response):
     API_KEY = 16
     API_VERSION = 0
     SCHEMA = Schema(
@@ -126,7 +126,7 @@ class ListGroupsResponse_v0(Struct):
     )
 
 
-class ListGroupsRequest_v0(Struct):
+class ListGroupsRequest_v0(Request):
     API_KEY = 16
     API_VERSION = 0
     RESPONSE_TYPE = ListGroupsResponse_v0
@@ -137,7 +137,7 @@ class ListGroupsRequest_v0(Struct):
 ListGroupsResponse = [ListGroupsResponse_v0]
 
 
-class DescribeGroupsResponse_v0(Struct):
+class DescribeGroupsResponse_v0(Response):
     API_KEY = 15
     API_VERSION = 0
     SCHEMA = Schema(
@@ -156,7 +156,7 @@ class DescribeGroupsResponse_v0(Struct):
     )
 
 
-class DescribeGroupsRequest_v0(Struct):
+class DescribeGroupsRequest_v0(Request):
     API_KEY = 15
     API_VERSION = 0
     RESPONSE_TYPE = DescribeGroupsResponse_v0
@@ -169,7 +169,7 @@ class DescribeGroupsRequest_v0(Struct):
 DescribeGroupsResponse = [DescribeGroupsResponse_v0]
 
 
-class SaslHandShakeResponse_v0(Struct):
+class SaslHandShakeResponse_v0(Response):
     API_KEY = 17
     API_VERSION = 0
     SCHEMA = Schema(
@@ -178,7 +178,7 @@ class SaslHandShakeResponse_v0(Struct):
     )
 
 
-class SaslHandShakeRequest_v0(Struct):
+class SaslHandShakeRequest_v0(Request):
     API_KEY = 17
     API_VERSION = 0
     RESPONSE_TYPE = SaslHandShakeResponse_v0

kafka/protocol/api.py

@@ -1,5 +1,7 @@
 from __future__ import absolute_import
 
+import abc
+
 from .struct import Struct
 from .types import Int16, Int32, String, Schema
 
@@ -16,3 +18,50 @@ def __init__(self, request, correlation_id=0, client_id='kafka-python'):
         super(RequestHeader, self).__init__(
             request.API_KEY, request.API_VERSION, correlation_id, client_id
         )
+
+
+class Request(Struct):
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractproperty
+    def API_KEY(self):
+        """Integer identifier for api request"""
+        pass
+
+    @abc.abstractproperty
+    def API_VERSION(self):
+        """Integer of api request version"""
+        pass
+
+    @abc.abstractproperty
+    def SCHEMA(self):
+        """An instance of Schema() representing the request structure"""
+        pass
+
+    @abc.abstractproperty
+    def RESPONSE_TYPE(self):
+        """The Response class associated with the api request"""
+        pass
+
+    def expect_response(self):
+        """Override this method if an api request does not always generate a response"""
+        return True
+
+
+class Response(Struct):
+    __metaclass__ = abc.ABCMeta
+
+    @abc.abstractproperty
+    def API_KEY(self):
+        """Integer identifier for api request/response"""
+        pass
+
+    @abc.abstractproperty
+    def API_VERSION(self):
+        """Integer of api request/response version"""
+        pass
+
+    @abc.abstractproperty
+    def SCHEMA(self):
+        """An instance of Schema() representing the response structure"""
+        pass

kafka/protocol/commit.py

@@ -1,10 +1,10 @@
 from __future__ import absolute_import
 
-from .struct import Struct
+from .api import Request, Response
 from .types import Array, Int16, Int32, Int64, Schema, String
 
 
-class OffsetCommitResponse_v0(Struct):
+class OffsetCommitResponse_v0(Response):
     API_KEY = 8
     API_VERSION = 0
     SCHEMA = Schema(
@@ -16,19 +16,19 @@ class OffsetCommitResponse_v0(Struct):
     )
 
 
-class OffsetCommitResponse_v1(Struct):
+class OffsetCommitResponse_v1(Response):
     API_KEY = 8
     API_VERSION = 1
     SCHEMA = OffsetCommitResponse_v0.SCHEMA
 
 
-class OffsetCommitResponse_v2(Struct):
+class OffsetCommitResponse_v2(Response):
     API_KEY = 8
     API_VERSION = 2
     SCHEMA = OffsetCommitResponse_v1.SCHEMA
 
 
-class OffsetCommitRequest_v0(Struct):
+class OffsetCommitRequest_v0(Request):
     API_KEY = 8
     API_VERSION = 0  # Zookeeper-backed storage
     RESPONSE_TYPE = OffsetCommitResponse_v0
@@ -43,7 +43,7 @@ class OffsetCommitRequest_v0(Struct):
     )
 
 
-class OffsetCommitRequest_v1(Struct):
+class OffsetCommitRequest_v1(Request):
     API_KEY = 8
     API_VERSION = 1  # Kafka-backed storage
     RESPONSE_TYPE = OffsetCommitResponse_v1
@@ -61,7 +61,7 @@ class OffsetCommitRequest_v1(Struct):
     )
 
 
-class OffsetCommitRequest_v2(Struct):
+class OffsetCommitRequest_v2(Request):
     API_KEY = 8
     API_VERSION = 2  # added retention_time, dropped timestamp
     RESPONSE_TYPE = OffsetCommitResponse_v2
@@ -87,7 +87,7 @@ class OffsetCommitRequest_v2(Struct):
                         OffsetCommitResponse_v2]
 
 
-class OffsetFetchResponse_v0(Struct):
+class OffsetFetchResponse_v0(Response):
     API_KEY = 9
     API_VERSION = 0
     SCHEMA = Schema(
@@ -101,13 +101,13 @@ class OffsetFetchResponse_v0(Struct):
     )
 
 
-class OffsetFetchResponse_v1(Struct):
+class OffsetFetchResponse_v1(Response):
     API_KEY = 9
     API_VERSION = 1
     SCHEMA = OffsetFetchResponse_v0.SCHEMA
 
 
-class OffsetFetchResponse_v2(Struct):
+class OffsetFetchResponse_v2(Response):
     # Added in KIP-88
     API_KEY = 9
     API_VERSION = 2
@@ -123,7 +123,7 @@ class OffsetFetchResponse_v2(Struct):
     )
 
 
-class OffsetFetchRequest_v0(Struct):
+class OffsetFetchRequest_v0(Request):
     API_KEY = 9
     API_VERSION = 0  # zookeeper-backed storage
     RESPONSE_TYPE = OffsetFetchResponse_v0
@@ -135,14 +135,14 @@ class OffsetFetchRequest_v0(Struct):
     )
 
 
-class OffsetFetchRequest_v1(Struct):
+class OffsetFetchRequest_v1(Request):
     API_KEY = 9
     API_VERSION = 1  # kafka-backed storage
     RESPONSE_TYPE = OffsetFetchResponse_v1
     SCHEMA = OffsetFetchRequest_v0.SCHEMA
 
 
-class OffsetFetchRequest_v2(Struct):
+class OffsetFetchRequest_v2(Request):
     # KIP-88: Allows passing null topics to return offsets for all partitions
     # that the consumer group has a stored offset for, even if no consumer in
     # the group is currently consuming that partition.
@@ -158,7 +158,7 @@ class OffsetFetchRequest_v2(Struct):
     OffsetFetchResponse_v2]
 
 
-class GroupCoordinatorResponse_v0(Struct):
+class GroupCoordinatorResponse_v0(Response):
     API_KEY = 10
     API_VERSION = 0
     SCHEMA = Schema(
@@ -169,7 +169,7 @@ class GroupCoordinatorResponse_v0(Struct):
     )
 
 
-class GroupCoordinatorRequest_v0(Struct):
+class GroupCoordinatorRequest_v0(Request):
     API_KEY = 10
     API_VERSION = 0
     RESPONSE_TYPE = GroupCoordinatorResponse_v0