Dynamic protocol classes using upstream json schemas (#2727)

Author: dpkp

kafka/protocol/new/__init__.py

@@ -0,0 +1,27 @@
+from .acl import *
+from .cluster import *
+from .groups import *
+from .topics import *
+
+__all__ = [
+    'CreateAclsRequest', 'CreateAclsResponse',
+    'DeleteAclsRequest', 'DeleteAclsResponse',
+    'DescribeAclsRequest', 'DescribeAclsResponse',
+    'ACLResourceType', 'ACLOperation',
+    'ACLPermissionType', 'ACLResourcePatternType',
+
+    'DescribeClusterRequest', 'DescribeClusterResponse',
+    'DescribeConfigsRequest', 'DescribeConfigsResponse',
+    'AlterConfigsRequest', 'AlterConfigsResponse',
+    'DescribeLogDirsRequest', 'DescribeLogDirsResponse',
+    'ElectLeadersRequest', 'ElectLeadersResponse', 'ElectionType',
+
+    'DescribeGroupsRequest', 'DescribeGroupsResponse',
+    'ListGroupsRequest', 'ListGroupsResponse',
+    'DeleteGroupsRequest', 'DeleteGroupsResponse',
+
+    'CreateTopicsRequest', 'CreateTopicsResponse',
+    'DeleteTopicsRequest', 'DeleteTopicsResponse',
+    'CreatePartitionsRequest', 'CreatePartitionsResponse',
+    'DeleteRecordsRequest', 'DeleteRecordsResponse',
+]

kafka/protocol/new/admin/__init__.py

@@ -0,0 +1,83 @@
+from enum import IntEnum
+
+from ..api_message import ApiMessage
+
+
+class CreateAclsRequest(ApiMessage): pass
+class CreateAclsResponse(ApiMessage): pass
+
+class DeleteAclsRequest(ApiMessage): pass
+class DeleteAclsResponse(ApiMessage): pass
+
+class DescribeAclsRequest(ApiMessage): pass
+class DescribeAclsResponse(ApiMessage): pass
+
+
+class ACLResourceType(IntEnum):
+    """Type of kafka resource to set ACL for
+
+    The ANY value is only valid in a filter context
+    """
+    UNKNOWN = 0,
+    ANY = 1,
+    CLUSTER = 4,
+    DELEGATION_TOKEN = 6,
+    GROUP = 3,
+    TOPIC = 2,
+    TRANSACTIONAL_ID = 5
+
+
+class ACLOperation(IntEnum):
+    """Type of operation
+
+    The ANY value is only valid in a filter context
+    """
+    UNKNOWN = 0,
+    ANY = 1,
+    ALL = 2,
+    READ = 3,
+    WRITE = 4,
+    CREATE = 5,
+    DELETE = 6,
+    ALTER = 7,
+    DESCRIBE = 8,
+    CLUSTER_ACTION = 9,
+    DESCRIBE_CONFIGS = 10,
+    ALTER_CONFIGS = 11,
+    IDEMPOTENT_WRITE = 12,
+    CREATE_TOKENS = 13,
+    DESCRIBE_TOKENS = 14
+
+
+class ACLPermissionType(IntEnum):
+    """An enumerated type of permissions
+
+    The ANY value is only valid in a filter context
+    """
+    UNKNOWN = 0,
+    ANY = 1,
+    DENY = 2,
+    ALLOW = 3
+
+
+class ACLResourcePatternType(IntEnum):
+    """An enumerated type of resource patterns
+
+    More details on the pattern types and how they work
+    can be found in KIP-290 (Support for prefixed ACLs)
+    https://cwiki.apache.org/confluence/display/KAFKA/KIP-290%3A+Support+for+Prefixed+ACLs
+    """
+    UNKNOWN = 0,
+    ANY = 1,
+    MATCH = 2,
+    LITERAL = 3,
+    PREFIXED = 4
+
+
+__all__ = [
+    'CreateAclsRequest', 'CreateAclsResponse',
+    'DeleteAclsRequest', 'DeleteAclsResponse',
+    'DescribeAclsRequest', 'DescribeAclsResponse',
+    'ACLResourceType', 'ACLOperation',
+    'ACLPermissionType', 'ACLResourcePatternType',
+]

kafka/protocol/new/admin/acl.py

@@ -0,0 +1,35 @@
+from enum import IntEnum
+
+from ..api_message import ApiMessage
+
+
+class DescribeClusterRequest(ApiMessage): pass
+class DescribeClusterResponse(ApiMessage): pass
+
+class DescribeConfigsRequest(ApiMessage): pass
+class DescribeConfigsResponse(ApiMessage): pass
+
+class AlterConfigsRequest(ApiMessage): pass
+class AlterConfigsResponse(ApiMessage): pass
+
+class DescribeLogDirsRequest(ApiMessage): pass
+class DescribeLogDirsResponse(ApiMessage): pass
+
+class ElectLeadersRequest(ApiMessage): pass
+class ElectLeadersResponse(ApiMessage): pass
+
+class ElectionType(IntEnum):
+    """ Leader election type
+    """
+
+    PREFERRED = 0,
+    UNCLEAN = 1
+
+
+__all__ = [
+    'DescribeClusterRequest', 'DescribeClusterResponse',
+    'DescribeConfigsRequest', 'DescribeConfigsResponse',
+    'AlterConfigsRequest', 'AlterConfigsResponse',
+    'DescribeLogDirsRequest', 'DescribeLogDirsResponse',
+    'ElectLeadersRequest', 'ElectLeadersResponse', 'ElectionType',
+]

kafka/protocol/new/admin/cluster.py

@@ -0,0 +1,18 @@
+from ..api_message import ApiMessage
+
+
+class DescribeGroupsRequest(ApiMessage): pass
+class DescribeGroupsResponse(ApiMessage): pass
+
+class ListGroupsRequest(ApiMessage): pass
+class ListGroupsResponse(ApiMessage): pass
+
+class DeleteGroupsRequest(ApiMessage): pass
+class DeleteGroupsResponse(ApiMessage): pass
+
+
+__all__ = [
+    'DescribeGroupsRequest', 'DescribeGroupsResponse',
+    'ListGroupsRequest', 'ListGroupsResponse',
+    'DeleteGroupsRequest', 'DeleteGroupsResponse',
+]

kafka/protocol/new/admin/groups.py

@@ -0,0 +1,22 @@
+from ..api_message import ApiMessage
+
+
+class CreateTopicsRequest(ApiMessage): pass
+class CreateTopicsResponse(ApiMessage): pass
+
+class DeleteTopicsRequest(ApiMessage): pass
+class DeleteTopicsResponse(ApiMessage): pass
+
+class CreatePartitionsRequest(ApiMessage): pass
+class CreatePartitionsResponse(ApiMessage): pass
+
+class DeleteRecordsRequest(ApiMessage): pass
+class DeleteRecordsResponse(ApiMessage): pass
+
+
+__all__ = [
+    'CreateTopicsRequest', 'CreateTopicsResponse',
+    'DeleteTopicsRequest', 'DeleteTopicsResponse',
+    'CreatePartitionsRequest', 'CreatePartitionsResponse',
+    'DeleteRecordsRequest', 'DeleteRecordsResponse',
+]

kafka/protocol/new/admin/topics.py

@@ -0,0 +1,59 @@
+from .data_container import DataContainer, SlotsBuilder
+from .schemas import BaseField, StructField, load_json
+
+
+class ApiHeaderMeta(SlotsBuilder):
+    def __new__(metacls, name, bases, attrs, **kw):
+        if kw.get('init', True):
+            json = load_json(name)
+            attrs['_json'] = json
+            attrs['_struct'] = StructField(json)
+        return super().__new__(metacls, name, bases, attrs, **kw)
+
+
+class ApiHeader(DataContainer, metaclass=ApiHeaderMeta, init=False):
+    __slots__ = ()
+
+    def __init_subclass__(cls, **kw):
+        super().__init_subclass__(**kw)
+        if kw.get('init', True):
+            # pylint: disable=E1101
+            assert cls._json['type'] == 'header'
+            cls._flexible_versions = BaseField.parse_versions(cls._json['flexibleVersions'])
+            cls._valid_versions = BaseField.parse_versions(cls._json['validVersions'])
+
+    def encode(self, flexible=False):
+        # Request versions are 1-2, Response versions are 0-1
+        version = self._flexible_versions[0] if flexible else self._valid_versions[0] # pylint: disable=E1136
+        # compact=False is probably wrong,
+        # but it works to make sure that the client_id request header field
+        # is never encoded as compact (required to support ApiVersionsRequest for unsupported version)
+        return super().encode(version=version, compact=False, tagged=flexible)
+
+    @classmethod
+    def decode(cls, data, flexible=False):
+        # Request versions are 1-2, Response versions are 0-1
+        version = cls._flexible_versions[0] if flexible else cls._valid_versions[0] # pylint: disable=E1136
+        return cls._struct.decode(data, version=version, compact=False, tagged=flexible, data_class=cls)
+
+
+class ResponseClassRegistry:
+    _response_class_registry = {}
+
+    @classmethod
+    def register_response_class(cls, response_class):
+        cls._response_class_registry[response_class.API_KEY] = response_class
+
+    @classmethod
+    def get_response_class(cls, request_header):
+        response_class = cls._response_class_registry.get(request_header.request_api_key)
+        if response_class is not None:
+            return response_class[request_header.request_api_version]
+
+
+class RequestHeader(ApiHeader):
+    def get_response_class(self):
+        return ResponseClassRegistry.get_response_class(self)
+
+
+class ResponseHeader(ApiHeader): pass