From 8a72f1a567c4d71dc6a412f3d41ad80f56c495ed Mon Sep 17 00:00:00 2001 From: Taras Voinarovskyi Date: Thu, 19 Mar 2020 20:15:21 +0200 Subject: Add API version determination code for newer kafka brokers --- kafka/conn.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/kafka/conn.py b/kafka/conn.py index c383123..797928d 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -26,6 +26,7 @@ from kafka.metrics.stats import Avg, Count, Max, Rate from kafka.oauth.abstract import AbstractTokenProvider from kafka.protocol.admin import SaslHandShakeRequest from kafka.protocol.commit import OffsetFetchRequest +from kafka.protocol.fetch import FetchRequest from kafka.protocol.metadata import MetadataRequest from kafka.protocol.parser import KafkaProtocol from kafka.protocol.types import Int32, Int8 @@ -1166,6 +1167,10 @@ class BrokerConnection(object): # in reverse order. As soon as we find one that works, return it test_cases = [ # format (, ) + ((2, 3, 0), FetchRequest[0].API_KEY, 11), + ((2, 1, 0), MetadataRequest[0].API_KEY, 7), + ((2, 0, 0), FetchRequest[0].API_KEY, 8), + ((1, 1, 0), FetchRequest[0].API_KEY, 7), ((1, 0, 0), MetadataRequest[5]), ((0, 11, 0), MetadataRequest[4]), ((0, 10, 2), OffsetFetchRequest[2]), -- cgit v1.2.1