Fixes to support integration testing with external KAFKA_URI (#2838)

Author: dpkp

test/integration/test_admin_integration.py

@@ -93,7 +93,7 @@ def test_create_describe_delete_acls(kafka_admin_client):
 def test_describe_configs_broker_resource_returns_configs(kafka_admin_client):
     """Tests that describe config returns configs for broker
     """
-    broker_id = kafka_admin_client._client.cluster._brokers[0].nodeId
+    broker_id = kafka_admin_client._client.least_loaded_node()
     configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
 
     assert len(configs) == 1
@@ -121,7 +121,7 @@ def test_describe_configs_topic_resource_returns_configs(topic, kafka_admin_clie
 def test_describe_configs_mixed_resources_returns_configs(topic, kafka_admin_client):
     """Tests that describe config returns configs for mixed resource types (topic + broker)
     """
-    broker_id = kafka_admin_client._client.cluster._brokers[0].nodeId
+    broker_id = kafka_admin_client._client.least_loaded_node()
     configs = kafka_admin_client.describe_configs([
         ConfigResource(ConfigResourceType.TOPIC, topic),
         ConfigResource(ConfigResourceType.BROKER, broker_id)])
@@ -146,14 +146,6 @@ def test_describe_configs_invalid_broker_id_raises(kafka_admin_client):
         kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
 
 
-@pytest.mark.skipif(env_kafka_version() < (0, 11), reason='Describe consumer group requires broker >=0.11')
-def test_describe_consumer_group_does_not_exist(kafka_admin_client):
-    """Tests that the describe consumer group call fails if the group coordinator is not available
-    """
-    with pytest.raises(CoordinatorNotAvailableError):
-        kafka_admin_client.describe_consumer_groups(['test'])
-
-
 @pytest.mark.skipif(env_kafka_version() < (0, 11), reason='Describe consumer group requires broker >=0.11')
 def test_describe_consumer_group_exists(kafka_admin_client, kafka_consumer_factory, topic):
     """Tests that the describe consumer group call returns valid consumer group information

test/integration/test_sasl_integration.py

@@ -1,4 +1,5 @@
 import logging
+import os
 import uuid
 import time
 
@@ -8,6 +9,7 @@
 from kafka.protocol.metadata import MetadataRequest_v1
 from test.testutil import assert_message_count, env_kafka_version, random_string, special_to_underscore
 
+pytestmark = pytest.mark.skipif("KAFKA_URI" in os.environ, reason="Testing on external Kafka Broker")
 
 @pytest.fixture(
     params=[

test/test_fetcher.py

@@ -929,7 +929,7 @@ def test_success_does_not_check_exception(self, client, mocker):
         def poll_side_effect(**kw):
             f = kw.get('future')
             if f is future1:
-                # Succeeds but has retry partitions — the bug was that code
+                # Succeeds but has retry partitions -- the bug was that code
                 # would fall through to check future.exception (which is None),
                 # causing an AttributeError
                 f.success(({tp0: offset0}, {tp1}))