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

1 files changed, 12 insertions, 3 deletions

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,15 +6,24 @@ 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"""
     consumer = kafka_consumer_factory(auto_offset_reset='earliest')