diff options
| author | Tincu Gabriel <gabi@aiven.io> | 2020-05-05 13:29:23 +0200 |
|---|---|---|
| committer | GitHub <noreply@github.com> | 2020-05-05 14:29:23 +0300 |
| commit | 6fc008137c75c751a9fbea3e0ef36d2870119c7b (patch) | |
| tree | 667c4cb56dd57c819e8a73387c6689383a5ac564 /kafka/conn.py | |
| parent | f9e0264e0b0f8d92afb6177d51976795e3bdbcd8 (diff) | |
| download | kafka-python-6fc008137c75c751a9fbea3e0ef36d2870119c7b.tar.gz | |
Add logic for inferring newer broker versions (#2038)
* Add logic for inferring newer broker versions
- New Fetch / ListOffsets request / response objects
- Add new test cases to inferr code based on mentioned objects
- Add unit test to check inferred version against whatever resides in KAFKA_VERSION
- Add new kafka broker versions to integration list
- Add more kafka broker versions to travis task list
- Add support for broker version 2.5 id
* Implement PR change requests: fewer versions for travis testing, remove unused older versions for inference code, remove one minor version from known server list
Do not use newly created ACL request / responses in allowed version lists, due to flexible versions enabling in kafka actually requiring a serialization protocol header update
Revert admin client file change
Diffstat (limited to 'kafka/conn.py')
| -rw-r--r-- | kafka/conn.py | 12 |
1 files changed, 11 insertions, 1 deletions
diff --git a/kafka/conn.py b/kafka/conn.py index c383123..5c72875 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -24,9 +24,12 @@ 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 +from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2 from kafka.protocol.commit import OffsetFetchRequest +from kafka.protocol.offset import OffsetRequest +from kafka.protocol.produce import ProduceRequest from kafka.protocol.metadata import MetadataRequest +from kafka.protocol.fetch import FetchRequest from kafka.protocol.parser import KafkaProtocol from kafka.protocol.types import Int32, Int8 from kafka.scram import ScramClient @@ -1166,6 +1169,13 @@ class BrokerConnection(object): # in reverse order. As soon as we find one that works, return it test_cases = [ # format (<broker version>, <needed struct>) + ((2, 5, 0), DescribeAclsRequest_v2), + ((2, 4, 0), ProduceRequest[8]), + ((2, 3, 0), FetchRequest[11]), + ((2, 2, 0), OffsetRequest[5]), + ((2, 1, 0), FetchRequest[10]), + ((2, 0, 0), FetchRequest[8]), + ((1, 1, 0), FetchRequest[7]), ((1, 0, 0), MetadataRequest[5]), ((0, 11, 0), MetadataRequest[4]), ((0, 10, 2), OffsetFetchRequest[2]), |