fix(confluent-kafka): skip recv spans when poll/consume returns no messages

Matches Java instrumentation behavior: only create a recv span when a
message is actually received, avoiding empty spans on idle polls.

Fixes #3590

Author: sterchelen

instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py

@@ -383,11 +383,11 @@ def wrap_poll(func, instance, tracer, args, kwargs):
         if instance._current_consume_span:
             _end_current_consume_span(instance)
 
-        with tracer.start_as_current_span(
-            "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER
-        ):
-            record = func(*args, **kwargs)
-            if record:
+        record = func(*args, **kwargs)
+        if record:
+            with tracer.start_as_current_span(
+                "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER
+            ):
                 _create_new_consume_span(instance, tracer, [record])
                 _enrich_span(
                     instance._current_consume_span,
@@ -396,9 +396,9 @@ def wrap_poll(func, instance, tracer, args, kwargs):
                     record.offset(),
                     operation=MessagingOperationTypeValues.PROCESS,
                 )
-        instance._current_context_token = context.attach(
-            trace.set_span_in_context(instance._current_consume_span)
-        )
+            instance._current_context_token = context.attach(
+                trace.set_span_in_context(instance._current_consume_span)
+            )
 
         return record
 
@@ -407,21 +407,20 @@ def wrap_consume(func, instance, tracer, args, kwargs):
         if instance._current_consume_span:
             _end_current_consume_span(instance)
 
-        with tracer.start_as_current_span(
-            "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER
-        ):
-            records = func(*args, **kwargs)
-            if len(records) > 0:
+        records = func(*args, **kwargs)
+        if len(records) > 0:
+            with tracer.start_as_current_span(
+                "recv", end_on_exit=True, kind=trace.SpanKind.CONSUMER
+            ):
                 _create_new_consume_span(instance, tracer, records)
                 _enrich_span(
                     instance._current_consume_span,
                     records[0].topic(),
                     operation=MessagingOperationTypeValues.PROCESS,
                 )
-
-        instance._current_context_token = context.attach(
-            trace.set_span_in_context(instance._current_consume_span)
-        )
+            instance._current_context_token = context.attach(
+                trace.set_span_in_context(instance._current_consume_span)
+            )
 
         return records
 

instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py

@@ -197,7 +197,6 @@ def test_poll(self) -> None:
                     MESSAGING_MESSAGE_ID: "topic-30.1.3",
                 },
             },
-            {"name": "recv", "attributes": {}},
         ]
 
         consumer = MockConsumer(
@@ -213,7 +212,7 @@ def test_poll(self) -> None:
         consumer.poll()
         consumer.poll()
         consumer.poll()
-        consumer.poll()
+        consumer.poll()  # empty poll — must not produce a span
 
         span_list = self.memory_exporter.get_finished_spans()
         self._compare_spans(span_list, expected_spans)
@@ -259,7 +258,6 @@ def test_consume(self) -> None:
                     SpanAttributes.MESSAGING_DESTINATION_KIND: MessagingDestinationKindValues.QUEUE.value,
                 },
             },
-            {"name": "recv", "attributes": {}},
         ]
 
         consumer = MockConsumer(
@@ -276,10 +274,46 @@ def test_consume(self) -> None:
         consumer.consume(3)
         consumer.consume(1)
         consumer.consume(2)
-        consumer.consume(1)
+        consumer.consume(1)  # empty consume — must not produce a span
         span_list = self.memory_exporter.get_finished_spans()
         self._compare_spans(span_list, expected_spans)
 
+    def test_poll_empty_does_not_create_span(self) -> None:
+        instrumentation = ConfluentKafkaInstrumentor()
+        consumer = MockConsumer(
+            [],
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            },
+        )
+        self.memory_exporter.clear()
+        consumer = instrumentation.instrument_consumer(consumer)
+        consumer.poll()
+        consumer.poll()
+
+        span_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 0)
+
+    def test_consume_empty_does_not_create_span(self) -> None:
+        instrumentation = ConfluentKafkaInstrumentor()
+        consumer = MockConsumer(
+            [],
+            {
+                "bootstrap.servers": "localhost:29092",
+                "group.id": "mygroup",
+                "auto.offset.reset": "earliest",
+            },
+        )
+        self.memory_exporter.clear()
+        consumer = instrumentation.instrument_consumer(consumer)
+        consumer.consume(5)
+        consumer.consume(5)
+
+        span_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 0)
+
     def test_close(self) -> None:
         instrumentation = ConfluentKafkaInstrumentor()
         mocked_messages = [