opentelemetry-instrumentation-kafka-python: wait for metadata

CHANGELOG.md

@@ -21,6 +21,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `opentelemetry-instrumentation-kafka-python`: wait for metadata
+  ([#1260](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1260))
 - `opentelemetry-instrumentation-boto3sqs` Make propagation compatible with other SQS instrumentations, add 'messaging.url' span attribute, and fix missing package dependencies.
   ([#1234](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/1234)) 
 - `opentelemetry-instrumentation-pymongo` Change span names to not contain queries but only database name and command name

instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py

@@ -2,6 +2,7 @@
 from logging import getLogger
 from typing import Callable, Dict, List, Optional
 
+import kafka.errors as KafkaErrors
 from kafka.record.abc import ABCRecord
 
 from opentelemetry import context, propagate, trace
@@ -146,6 +147,12 @@ def _traced_send(func, instance, args, kwargs):
             kwargs["headers"] = headers
 
         topic = KafkaPropertiesExtractor.extract_send_topic(args, kwargs)
+        try:
+            instance._wait_on_metadata(
+                topic, instance.config["max_block_ms"] / 1000.0
+            )
+        except KafkaErrors.BrokerResponseError as kafka_exception:
+            _LOG.exception(kafka_exception)
         bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(
             instance
         )