Use wait_for_metadata=True when creating topic fixtures with admin client (#2858)

Author: dpkp

test/integration/fixtures.py

@@ -42,11 +42,11 @@ def _create_topics_via_admin(broker, topic_names, num_partitions, replication_fa
     admin = KafkaAdminClient(**params)
     try:
         topics = [NewTopic(name, num_partitions, replication_factor) for name in topic_names]
-        admin.create_topics(topics)
+        admin.create_topics(topics, wait_for_metadata=True)
     except InvalidReplicationFactorError:
         time.sleep(0.5)
         topics = [NewTopic(name, num_partitions, replication_factor) for name in topic_names]
-        admin.create_topics(topics)
+        admin.create_topics(topics, wait_for_metadata=True)
     finally:
         admin.close()
 

test/integration/test_consumer_integration.py

@@ -23,6 +23,8 @@ def test_consumer(consumer):
 
 
 def test_consumer_topics(consumer, topic):
+    # The `topic` fixture waits for the topic to be visible in broker
+    # metadata before returning, so a single poll + fetch is sufficient here.
     consumer.poll(timeout_ms=500)
     assert topic in consumer.topics()
     assert len(consumer.partitions_for_topic(topic)) > 0