diff options
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/conn.py | 3 | ||||
| -rw-r--r-- | kafka/protocol/__init__.py | 1 | ||||
| -rw-r--r-- | kafka/protocol/admin.py | 42 | ||||
| -rw-r--r-- | kafka/protocol/types.py | 13 |
4 files changed, 57 insertions, 2 deletions
diff --git a/kafka/conn.py b/kafka/conn.py index 5c72875..cac3548 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -24,7 +24,7 @@ import kafka.errors as Errors from kafka.future import Future from kafka.metrics.stats import Avg, Count, Max, Rate from kafka.oauth.abstract import AbstractTokenProvider -from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2 +from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2, DescribeClientQuotasRequest from kafka.protocol.commit import OffsetFetchRequest from kafka.protocol.offset import OffsetRequest from kafka.protocol.produce import ProduceRequest @@ -1169,6 +1169,7 @@ class BrokerConnection(object): # in reverse order. As soon as we find one that works, return it test_cases = [ # format (<broker version>, <needed struct>) + ((2, 6, 0), DescribeClientQuotasRequest[0]), ((2, 5, 0), DescribeAclsRequest_v2), ((2, 4, 0), ProduceRequest[8]), ((2, 3, 0), FetchRequest[11]), diff --git a/kafka/protocol/__init__.py b/kafka/protocol/__init__.py index 26dcc78..e739b5c 100644 --- a/kafka/protocol/__init__.py +++ b/kafka/protocol/__init__.py @@ -43,4 +43,5 @@ API_KEYS = { 40: 'ExpireDelegationToken', 41: 'DescribeDelegationToken', 42: 'DeleteGroups', + 48: 'DescribeClientQuotas', } diff --git a/kafka/protocol/admin.py b/kafka/protocol/admin.py index f3b691a..63a3327 100644 --- a/kafka/protocol/admin.py +++ b/kafka/protocol/admin.py @@ -1,7 +1,7 @@ from __future__ import absolute_import from kafka.protocol.api import Request, Response -from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String +from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String, Float64 class ApiVersionResponse_v0(Response): @@ -923,3 +923,43 @@ DeleteGroupsRequest = [ DeleteGroupsResponse = [ DeleteGroupsResponse_v0, DeleteGroupsResponse_v1 ] + + +class DescribeClientQuotasResponse_v0(Request): + API_KEY = 48 + API_VERSION = 0 + SCHEMA = Schema( + ('throttle_time_ms', Int32), + ('error_code', Int16), + ('error_message', String('utf-8')), + ('entries', Array( + ('entity', Array( + ('entity_type', String('utf-8')), + ('entity_name', String('utf-8')))), + ('values', Array( + ('name', String('utf-8')), + ('value', Float64))))), + ) + + +class DescribeClientQuotasRequest_v0(Request): + API_KEY = 48 + API_VERSION = 0 + RESPONSE_TYPE = DescribeClientQuotasResponse_v0 + SCHEMA = Schema( + ('components', Array( + ('entity_type', String('utf-8')), + ('match_type', Int8), + ('match', String('utf-8')), + )), + ('strict', Boolean) + ) + + +DescribeClientQuotasRequest = [ + DescribeClientQuotasRequest_v0, +] + +DescribeClientQuotasResponse = [ + DescribeClientQuotasResponse_v0, +] diff --git a/kafka/protocol/types.py b/kafka/protocol/types.py index d508b26..ade1bc6 100644 --- a/kafka/protocol/types.py +++ b/kafka/protocol/types.py @@ -77,6 +77,19 @@ class Int64(AbstractType): return _unpack(cls._unpack, data.read(8)) +class Float64(AbstractType): + _pack = struct.Struct('>d').pack + _unpack = struct.Struct('>d').unpack + + @classmethod + def encode(cls, value): + return _pack(cls._pack, value) + + @classmethod + def decode(cls, data): + return _unpack(cls._unpack, data.read(8)) + + class String(AbstractType): def __init__(self, encoding='utf-8'): self.encoding = encoding |