fix: sync span_kind with spec

Author: dimastbk

instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py

@@ -253,7 +253,7 @@ def _enrich_send_span(
     span.set_attribute(messaging_attributes.MESSAGING_OPERATION_NAME, "send")
     span.set_attribute(
         messaging_attributes.MESSAGING_OPERATION_TYPE,
-        messaging_attributes.MessagingOperationTypeValues.PUBLISH.value,
+        messaging_attributes.MessagingOperationTypeValues.SEND.value,
     )
 
 
@@ -444,7 +444,7 @@ async def _create_consumer_span(
     with tracer.start_as_current_span(
         span_name,
         context=extracted_context,
-        kind=trace.SpanKind.CONSUMER,
+        kind=trace.SpanKind.CLIENT,
     ) as span:
         new_context = trace.set_span_in_context(span, extracted_context)
         token = context.attach(new_context)

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py

@@ -158,7 +158,7 @@ async def test_getone(self) -> None:
         expected_spans = [
             {
                 "name": "topic_1 receive",
-                "kind": SpanKind.CONSUMER,
+                "kind": SpanKind.CLIENT,
                 "attributes": {
                     messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value,
                     server_attributes.SERVER_ADDRESS: "localhost",
@@ -175,7 +175,7 @@ async def test_getone(self) -> None:
             },
             {
                 "name": "topic_2 receive",
-                "kind": SpanKind.CONSUMER,
+                "kind": SpanKind.CLIENT,
                 "attributes": {
                     messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value,
                     server_attributes.SERVER_ADDRESS: "localhost",
@@ -293,7 +293,7 @@ async def test_getmany(self) -> None:
         expected_spans = [
             {
                 "name": "topic_1 receive",
-                "kind": SpanKind.CONSUMER,
+                "kind": SpanKind.CLIENT,
                 "attributes": {
                     messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value,
                     server_attributes.SERVER_ADDRESS: "localhost",
@@ -325,7 +325,7 @@ async def test_getmany(self) -> None:
             },
             {
                 "name": "topic_2 receive",
-                "kind": SpanKind.CONSUMER,
+                "kind": SpanKind.CLIENT,
                 "attributes": {
                     messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value,
                     server_attributes.SERVER_ADDRESS: "localhost",

instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py

@@ -347,7 +347,7 @@ async def test_create_consumer_span(
         tracer.start_as_current_span.assert_called_once_with(
             expected_span_name,
             context=extracted_context,
-            kind=SpanKind.CONSUMER,
+            kind=SpanKind.CLIENT,
         )
         span = tracer.start_as_current_span.return_value.__enter__()
         set_span_in_context.assert_called_once_with(span, extracted_context)