opentelemetry-instrumentation-kafka-python: wait for metadata

CHANGELOG.md

@@ -106,6 +106,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

@@ -75,9 +75,9 @@ def extract_send_partition(instance, args, kwargs):
             ):
                 return None
 
-            all_partitions = instance._metadata.partitions_for_topic(topic)
-            if all_partitions is None or len(all_partitions) == 0:
-                return None
+            instance._wait_on_metadata(
+                topic, instance.config["max_block_ms"] / 1000.0
+            )
 
             return instance._partition(
                 topic, partition, key, value, key_bytes, value_bytes

instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py

@@ -1,6 +1,7 @@
 from unittest import TestCase, mock
 
 from opentelemetry.instrumentation.kafka.utils import (
+    KafkaPropertiesExtractor,
     _create_consumer_span,
     _get_span_name,
     _kafka_getter,
@@ -208,3 +209,28 @@ def test_create_consumer_span(
             span, record, self.args, self.kwargs
         )
         detach.assert_called_once_with(attach.return_value)
+
+    @mock.patch(
+        "opentelemetry.instrumentation.kafka.utils.KafkaPropertiesExtractor"
+    )
+    def test_kafka_properties_extractor(
+        self,
+        kafka_properties_extractor: mock.MagicMock,
+    ):
+        kafka_properties_extractor._serialize.return_value = None
+        kafka_properties_extractor._partition.return_value = "partition"
+        assert (
+            KafkaPropertiesExtractor.extract_send_partition(
+                kafka_properties_extractor, self.args, self.kwargs
+            )
+            == "partition"
+        )
+        kafka_properties_extractor._wait_on_metadata.side_effect = Exception(
+            "mocked error"
+        )
+        assert (
+            KafkaPropertiesExtractor.extract_send_partition(
+                kafka_properties_extractor, self.args, self.kwargs
+            )
+            is None
+        )