Skip to content

Commit 8b8a48b

Browse files
Add Kafka 2.6.0 to tests and protocol compatibility matrix (#2162)
* Co-authored-by: Andrew Brown <[email protected]> * Co-authored-by: Aaron Brady <[email protected]>
1 parent 6c87155 commit 8b8a48b

File tree

12 files changed

+261
-8
lines changed

12 files changed

+261
-8
lines changed

.travis.yml

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ env:
1717
- KAFKA_VERSION=1.1.1
1818
- KAFKA_VERSION=2.4.0
1919
- KAFKA_VERSION=2.5.0
20+
- KAFKA_VERSION=2.6.0
2021

2122
addons:
2223
apt:

README.rst

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
Kafka Python client
22
------------------------
33

4-
.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
4+
.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
55
:target: https://kafka-python.readthedocs.io/en/master/compatibility.html
66
.. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
77
:target: https://pypi.python.org/pypi/kafka-python
@@ -158,4 +158,4 @@ for interacting with kafka brokers via the python repl. This is useful for
158158
testing, probing, and general experimentation. The protocol support is
159159
leveraged to enable a KafkaClient.check_version() method that
160160
probes a kafka broker and attempts to identify which version it is running
161-
(0.8.0 to 2.4+).
161+
(0.8.0 to 2.6+).

docs/compatibility.rst

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,12 @@
11
Compatibility
22
-------------
33

4-
.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
4+
.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
55
:target: https://kafka-python.readthedocs.io/compatibility.html
66
.. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
77
:target: https://pypi.python.org/pypi/kafka-python
88

9-
kafka-python is compatible with (and tested against) broker versions 2.5
9+
kafka-python is compatible with (and tested against) broker versions 2.6
1010
through 0.8.0 . kafka-python is not compatible with the 0.8.2-beta release.
1111

1212
Because the kafka server protocol is backwards compatible, kafka-python is

docs/index.rst

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
kafka-python
22
############
33

4-
.. image:: https://img.shields.io/badge/kafka-2.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
4+
.. image:: https://img.shields.io/badge/kafka-2.6%2C%202.5%2C%202.4%2C%202.3%2C%202.2%2C%202.1%2C%202.0%2C%201.1%2C%201.0%2C%200.11%2C%200.10%2C%200.9%2C%200.8-brightgreen.svg
55
:target: https://kafka-python.readthedocs.io/compatibility.html
66
.. image:: https://img.shields.io/pypi/pyversions/kafka-python.svg
77
:target: https://pypi.python.org/pypi/kafka-python
@@ -137,7 +137,7 @@ for interacting with kafka brokers via the python repl. This is useful for
137137
testing, probing, and general experimentation. The protocol support is
138138
leveraged to enable a :meth:`~kafka.KafkaClient.check_version()`
139139
method that probes a kafka broker and
140-
attempts to identify which version it is running (0.8.0 to 2.4+).
140+
attempts to identify which version it is running (0.8.0 to 2.6+).
141141

142142

143143
.. toctree::

kafka/conn.py

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@
2424
from kafka.future import Future
2525
from kafka.metrics.stats import Avg, Count, Max, Rate
2626
from kafka.oauth.abstract import AbstractTokenProvider
27-
from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2
27+
from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2, DescribeClientQuotasRequest
2828
from kafka.protocol.commit import OffsetFetchRequest
2929
from kafka.protocol.offset import OffsetRequest
3030
from kafka.protocol.produce import ProduceRequest
@@ -1169,6 +1169,7 @@ def _infer_broker_version_from_api_versions(self, api_versions):
11691169
# in reverse order. As soon as we find one that works, return it
11701170
test_cases = [
11711171
# format (<broker version>, <needed struct>)
1172+
((2, 6, 0), DescribeClientQuotasRequest[0]),
11721173
((2, 5, 0), DescribeAclsRequest_v2),
11731174
((2, 4, 0), ProduceRequest[8]),
11741175
((2, 3, 0), FetchRequest[11]),

kafka/protocol/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,4 +43,5 @@
4343
40: 'ExpireDelegationToken',
4444
41: 'DescribeDelegationToken',
4545
42: 'DeleteGroups',
46+
48: 'DescribeClientQuotas',
4647
}

kafka/protocol/admin.py

Lines changed: 41 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
from __future__ import absolute_import
22

33
from kafka.protocol.api import Request, Response
4-
from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String
4+
from kafka.protocol.types import Array, Boolean, Bytes, Int8, Int16, Int32, Int64, Schema, String, Float64
55

66

77
class ApiVersionResponse_v0(Response):
@@ -923,3 +923,43 @@ class DeleteGroupsRequest_v1(Request):
923923
DeleteGroupsResponse = [
924924
DeleteGroupsResponse_v0, DeleteGroupsResponse_v1
925925
]
926+
927+
928+
class DescribeClientQuotasResponse_v0(Request):
929+
API_KEY = 48
930+
API_VERSION = 0
931+
SCHEMA = Schema(
932+
('throttle_time_ms', Int32),
933+
('error_code', Int16),
934+
('error_message', String('utf-8')),
935+
('entries', Array(
936+
('entity', Array(
937+
('entity_type', String('utf-8')),
938+
('entity_name', String('utf-8')))),
939+
('values', Array(
940+
('name', String('utf-8')),
941+
('value', Float64))))),
942+
)
943+
944+
945+
class DescribeClientQuotasRequest_v0(Request):
946+
API_KEY = 48
947+
API_VERSION = 0
948+
RESPONSE_TYPE = DescribeClientQuotasResponse_v0
949+
SCHEMA = Schema(
950+
('components', Array(
951+
('entity_type', String('utf-8')),
952+
('match_type', Int8),
953+
('match', String('utf-8')),
954+
)),
955+
('strict', Boolean)
956+
)
957+
958+
959+
DescribeClientQuotasRequest = [
960+
DescribeClientQuotasRequest_v0,
961+
]
962+
963+
DescribeClientQuotasResponse = [
964+
DescribeClientQuotasResponse_v0,
965+
]

kafka/protocol/types.py

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,19 @@ def decode(cls, data):
7777
return _unpack(cls._unpack, data.read(8))
7878

7979

80+
class Float64(AbstractType):
81+
_pack = struct.Struct('>d').pack
82+
_unpack = struct.Struct('>d').unpack
83+
84+
@classmethod
85+
def encode(cls, value):
86+
return _pack(cls._pack, value)
87+
88+
@classmethod
89+
def decode(cls, data):
90+
return _unpack(cls._unpack, data.read(8))
91+
92+
8093
class String(AbstractType):
8194
def __init__(self, encoding='utf-8'):
8295
self.encoding = encoding
Lines changed: 147 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,147 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one or more
2+
# contributor license agreements. See the NOTICE file distributed with
3+
# this work for additional information regarding copyright ownership.
4+
# The ASF licenses this file to You under the Apache License, Version 2.0
5+
# (the "License"); you may not use this file except in compliance with
6+
# the License. You may obtain a copy of the License at
7+
#
8+
# http://www.apache.org/licenses/LICENSE-2.0
9+
#
10+
# Unless required by applicable law or agreed to in writing, software
11+
# distributed under the License is distributed on an "AS IS" BASIS,
12+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
# See the License for the specific language governing permissions and
14+
# limitations under the License.
15+
# see kafka.server.KafkaConfig for additional details and defaults
16+
17+
############################# Server Basics #############################
18+
19+
# The id of the broker. This must be set to a unique integer for each broker.
20+
broker.id={broker_id}
21+
22+
############################# Socket Server Settings #############################
23+
24+
listeners={transport}://{host}:{port}
25+
security.inter.broker.protocol={transport}
26+
27+
{sasl_config}
28+
29+
ssl.keystore.location={ssl_dir}/kafka.server.keystore.jks
30+
ssl.keystore.password=foobar
31+
ssl.key.password=foobar
32+
ssl.truststore.location={ssl_dir}/kafka.server.truststore.jks
33+
ssl.truststore.password=foobar
34+
35+
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
36+
allow.everyone.if.no.acl.found=true
37+
38+
# The port the socket server listens on
39+
#port=9092
40+
41+
# Hostname the broker will bind to. If not set, the server will bind to all interfaces
42+
#host.name=localhost
43+
44+
# Hostname the broker will advertise to producers and consumers. If not set, it uses the
45+
# value for "host.name" if configured. Otherwise, it will use the value returned from
46+
# java.net.InetAddress.getCanonicalHostName().
47+
#advertised.host.name=<hostname routable by clients>
48+
49+
# The port to publish to ZooKeeper for clients to use. If this is not set,
50+
# it will publish the same port that the broker binds to.
51+
#advertised.port=<port accessible by clients>
52+
53+
# The number of threads handling network requests
54+
num.network.threads=3
55+
56+
# The number of threads doing disk I/O
57+
num.io.threads=8
58+
59+
# The send buffer (SO_SNDBUF) used by the socket server
60+
socket.send.buffer.bytes=102400
61+
62+
# The receive buffer (SO_RCVBUF) used by the socket server
63+
socket.receive.buffer.bytes=102400
64+
65+
# The maximum size of a request that the socket server will accept (protection against OOM)
66+
socket.request.max.bytes=104857600
67+
68+
69+
############################# Log Basics #############################
70+
71+
# A comma seperated list of directories under which to store log files
72+
log.dirs={tmp_dir}/data
73+
74+
# The default number of log partitions per topic. More partitions allow greater
75+
# parallelism for consumption, but this will also result in more files across
76+
# the brokers.
77+
num.partitions={partitions}
78+
default.replication.factor={replicas}
79+
80+
## Short Replica Lag -- Drops failed brokers out of ISR
81+
replica.lag.time.max.ms=1000
82+
replica.socket.timeout.ms=1000
83+
84+
############################# Log Flush Policy #############################
85+
86+
# Messages are immediately written to the filesystem but by default we only fsync() to sync
87+
# the OS cache lazily. The following configurations control the flush of data to disk.
88+
# There are a few important trade-offs here:
89+
# 1. Durability: Unflushed data may be lost if you are not using replication.
90+
# 2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
91+
# 3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
92+
# The settings below allow one to configure the flush policy to flush data after a period of time or
93+
# every N messages (or both). This can be done globally and overridden on a per-topic basis.
94+
95+
# The number of messages to accept before forcing a flush of data to disk
96+
#log.flush.interval.messages=10000
97+
98+
# The maximum amount of time a message can sit in a log before we force a flush
99+
#log.flush.interval.ms=1000
100+
101+
############################# Log Retention Policy #############################
102+
103+
# The following configurations control the disposal of log segments. The policy can
104+
# be set to delete segments after a period of time, or after a given size has accumulated.
105+
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
106+
# from the end of the log.
107+
108+
# The minimum age of a log file to be eligible for deletion
109+
log.retention.hours=168
110+
111+
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
112+
# segments don't drop below log.retention.bytes.
113+
#log.retention.bytes=1073741824
114+
115+
# The maximum size of a log segment file. When this size is reached a new log segment will be created.
116+
log.segment.bytes=1073741824
117+
118+
# The interval at which log segments are checked to see if they can be deleted according
119+
# to the retention policies
120+
log.retention.check.interval.ms=300000
121+
122+
# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.
123+
# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.
124+
log.cleaner.enable=false
125+
126+
# tune down offset topics to reduce setup time in tests
127+
offsets.commit.timeout.ms=500
128+
offsets.topic.num.partitions=2
129+
offsets.topic.replication.factor=1
130+
131+
# Allow shorter session timeouts for tests
132+
group.min.session.timeout.ms=1000
133+
134+
135+
############################# Zookeeper #############################
136+
137+
# Zookeeper connection string (see zookeeper docs for details).
138+
# This is a comma separated host:port pairs, each corresponding to a zk
139+
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
140+
# You can also append an optional chroot string to the urls to specify the
141+
# root directory for all kafka znodes.
142+
zookeeper.connect={zk_host}:{zk_port}/{zk_chroot}
143+
144+
# Timeout in ms for connecting to zookeeper
145+
zookeeper.connection.timeout.ms=30000
146+
# We want to expire kafka broker sessions quickly when brokers die b/c we restart them quickly
147+
zookeeper.session.timeout.ms=500
Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
KafkaServer {{
2+
{jaas_config}
3+
}};
4+
Client {{}};
Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one or more
2+
# contributor license agreements. See the NOTICE file distributed with
3+
# this work for additional information regarding copyright ownership.
4+
# The ASF licenses this file to You under the Apache License, Version 2.0
5+
# (the "License"); you may not use this file except in compliance with
6+
# the License. You may obtain a copy of the License at
7+
#
8+
# http://www.apache.org/licenses/LICENSE-2.0
9+
#
10+
# Unless required by applicable law or agreed to in writing, software
11+
# distributed under the License is distributed on an "AS IS" BASIS,
12+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
# See the License for the specific language governing permissions and
14+
# limitations under the License.
15+
16+
log4j.rootLogger=INFO, stdout, logfile
17+
18+
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
19+
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
20+
log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c)%n
21+
22+
log4j.appender.logfile=org.apache.log4j.FileAppender
23+
log4j.appender.logfile.File=${kafka.logs.dir}/server.log
24+
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
25+
log4j.appender.logfile.layout.ConversionPattern=[%d] %p %m (%c)%n
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one or more
2+
# contributor license agreements. See the NOTICE file distributed with
3+
# this work for additional information regarding copyright ownership.
4+
# The ASF licenses this file to You under the Apache License, Version 2.0
5+
# (the "License"); you may not use this file except in compliance with
6+
# the License. You may obtain a copy of the License at
7+
#
8+
# http://www.apache.org/licenses/LICENSE-2.0
9+
#
10+
# Unless required by applicable law or agreed to in writing, software
11+
# distributed under the License is distributed on an "AS IS" BASIS,
12+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
# See the License for the specific language governing permissions and
14+
# limitations under the License.
15+
# the directory where the snapshot is stored.
16+
dataDir={tmp_dir}
17+
# the port at which the clients will connect
18+
clientPort={port}
19+
clientPortAddress={host}
20+
# disable the per-ip limit on the number of connections since this is a non-production config
21+
maxClientCnxns=0

0 commit comments

Comments
 (0)