From 6fc008137c75c751a9fbea3e0ef36d2870119c7b Mon Sep 17 00:00:00 2001 From: Tincu Gabriel Date: Tue, 5 May 2020 13:29:23 +0200 Subject: 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 --- test/test_consumer_integration.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) (limited to 'test/test_consumer_integration.py') diff --git a/test/test_consumer_integration.py b/test/test_consumer_integration.py index 6e6bc94..90b7ed2 100644 --- a/test/test_consumer_integration.py +++ b/test/test_consumer_integration.py @@ -6,14 +6,23 @@ import pytest from kafka.vendor.six.moves import range import kafka.codec -from kafka.errors import ( - KafkaTimeoutError, UnsupportedCodecError, UnsupportedVersionError -) +from kafka.errors import UnsupportedCodecError, UnsupportedVersionError from kafka.structs import TopicPartition, OffsetAndTimestamp from test.testutil import Timer, assert_message_count, env_kafka_version, random_string +@pytest.mark.skipif(not env_kafka_version(), reason="No KAFKA_VERSION set") +def test_kafka_version_infer(kafka_consumer_factory): + consumer = kafka_consumer_factory() + actual_ver_major_minor = env_kafka_version()[:2] + client = consumer._client + conn = list(client._conns.values())[0] + inferred_ver_major_minor = conn.check_version()[:2] + assert actual_ver_major_minor == inferred_ver_major_minor, \ + "Was expecting inferred broker version to be %s but was %s" % (actual_ver_major_minor, inferred_ver_major_minor) + + @pytest.mark.skipif(not env_kafka_version(), reason="No KAFKA_VERSION set") def test_kafka_consumer(kafka_consumer_factory, send_messages): """Test KafkaConsumer""" -- cgit v1.2.1