Always set messaging operation (open-telemetry#9791)

Abhishekkr3003 · Nov 7, 2023 · 2fcddfe · 2fcddfe
1 parent 228d0cb
commit 2fcddfe
Show file tree

Hide file tree

Showing 31 changed files with 78 additions and 18 deletions.
diff --git a/...pentelemetry/instrumentation/api/instrumenter/messaging/MessagingAttributesExtractor.java b/...pentelemetry/instrumentation/api/instrumenter/messaging/MessagingAttributesExtractor.java
@@ -89,7 +89,7 @@ public void onStart(AttributesBuilder attributes, Context parentContext, REQUEST
         attributes,
         SemanticAttributes.MESSAGING_MESSAGE_PAYLOAD_COMPRESSED_SIZE_BYTES,
         getter.getMessagePayloadCompressedSize(request));
-    if (operation == RECEIVE || operation == PROCESS) {
+    if (operation != null) {
       internalSet(attributes, SemanticAttributes.MESSAGING_OPERATION, operation.operationName());
     }
   }
@@ -120,6 +120,10 @@ public void onEnd(
    */
   @Override
   public SpanKey internalGetSpanKey() {
+    if (operation == null) {
+      return null;
+    }
+
     switch (operation) {
       case PUBLISH:
         return SpanKey.PRODUCER;

diff --git a/...elemetry/instrumentation/api/instrumenter/messaging/MessagingAttributesExtractorTest.java b/...elemetry/instrumentation/api/instrumenter/messaging/MessagingAttributesExtractorTest.java
@@ -92,7 +92,7 @@ static Stream<Arguments> destinations() {
   void shouldExtractNoAttributesIfNoneAreAvailable() {
     // given
     AttributesExtractor<Map<String, String>, String> underTest =
-        MessagingAttributesExtractor.create(TestGetter.INSTANCE, MessageOperation.PUBLISH);
+        MessagingAttributesExtractor.create(TestGetter.INSTANCE, null);
 
     Context context = Context.root();
 

diff --git a/...c/main/groovy/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractSqsTracingTest.groovy b/...c/main/groovy/io/opentelemetry/instrumentation/awssdk/v1_11/AbstractSqsTracingTest.groovy
@@ -104,6 +104,7 @@ abstract class AbstractSqsTracingTest extends InstrumentationSpecification {
             "net.peer.port" sqsPort
             "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "testSdkSqs"
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.NET_PROTOCOL_NAME" "http"
             "$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
             "$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" Long
@@ -193,6 +194,7 @@ abstract class AbstractSqsTracingTest extends InstrumentationSpecification {
             "net.peer.port" sqsPort
             "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "testSdkSqs"
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.NET_PROTOCOL_NAME" "http"
             "$SemanticAttributes.NET_PROTOCOL_VERSION" "1.1"
             "$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" Long

diff --git a/...rc/main/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2ClientTest.groovy b/...rc/main/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2ClientTest.groovy
@@ -115,6 +115,7 @@ abstract class AbstractAws2ClientTest extends AbstractAws2ClientCoreTest {
             } else if (service == "Sqs" && operation == "SendMessage") {
               "aws.queue.url" QUEUE_URL
               "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "somequeue"
+              "$SemanticAttributes.MESSAGING_OPERATION" "publish"
               "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
               "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             } else if (service == "Kinesis") {
@@ -217,6 +218,7 @@ abstract class AbstractAws2ClientTest extends AbstractAws2ClientCoreTest {
             } else if (service == "Sqs" && operation == "SendMessage") {
               "aws.queue.url" QUEUE_URL
               "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "somequeue"
+              "$SemanticAttributes.MESSAGING_OPERATION" "publish"
               "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
               "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             } else if (service == "Kinesis") {

diff --git a/...ain/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2SqsTracingTest.groovy b/...ain/groovy/io/opentelemetry/instrumentation/awssdk/v2_2/AbstractAws2SqsTracingTest.groovy
@@ -154,6 +154,7 @@ abstract class AbstractAws2SqsTracingTest extends InstrumentationSpecification {
             "net.peer.port" sqsPort
             "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "testSdkSqs"
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
             "$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
             "$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
@@ -321,6 +322,7 @@ abstract class AbstractAws2SqsTracingTest extends InstrumentationSpecification {
             "net.peer.port" sqsPort
             "$SemanticAttributes.MESSAGING_SYSTEM" "AmazonSQS"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" "testSdkSqs"
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.HTTP_REQUEST_CONTENT_LENGTH" { it == null || it instanceof Long }
             "$SemanticAttributes.HTTP_RESPONSE_CONTENT_LENGTH" { it == null || it instanceof Long }
           }

diff --git a/...st/java/io/opentelemetry/javaagent/instrumentation/apachecamel/aws/AwsSpanAssertions.java b/...st/java/io/opentelemetry/javaagent/instrumentation/apachecamel/aws/AwsSpanAssertions.java
@@ -102,6 +102,8 @@ static SpanDataAssert sqs(
               equalTo(SemanticAttributes.MESSAGING_SYSTEM, "AmazonSQS")));
       if (spanName.endsWith("receive")) {
         attributeAssertions.add(equalTo(SemanticAttributes.MESSAGING_OPERATION, "receive"));
+      } else if (spanName.endsWith("publish")) {
+        attributeAssertions.add(equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"));
       }
     }
 

diff --git a/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms2InstrumentationTest.java b/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms2InstrumentationTest.java
@@ -169,6 +169,7 @@ void testMessageConsumer(
                       .hasAttributesSatisfyingExactly(
                           equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                           equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                          equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                           equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                           messagingTempDestination(isTemporary)));
 
@@ -231,6 +232,7 @@ void testMessageListener(
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary)),
                 span ->

diff --git a/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms1InstrumentationTest.java b/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v1_1/Jms1InstrumentationTest.java
@@ -129,6 +129,7 @@ void testMessageConsumer(
                       .hasAttributesSatisfyingExactly(
                           equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                           equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                          equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                           equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                           messagingTempDestination(isTemporary)));
 
@@ -191,6 +192,7 @@ void testMessageListener(
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary)),
                 span ->
@@ -269,6 +271,7 @@ void shouldCaptureMessageHeaders(
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary),
                             equalTo(
@@ -338,6 +341,7 @@ void shouldFailWhenSendingReadOnlyMessage(
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "jms"),
                             equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, destinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary))),
         trace ->

diff --git a/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/Jms3InstrumentationTest.java b/...est/java/io/opentelemetry/javaagent/instrumentation/jms/v3_0/Jms3InstrumentationTest.java
@@ -147,6 +147,7 @@ void testMessageConsumer(DestinationFactory destinationFactory, boolean isTempor
                           equalTo(
                               SemanticAttributes.MESSAGING_DESTINATION_NAME,
                               producerDestinationName),
+                          equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                           equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                           messagingTempDestination(isTemporary)));
 
@@ -214,6 +215,7 @@ void testMessageListener(DestinationFactory destinationFactory, boolean isTempor
                             equalTo(
                                 SemanticAttributes.MESSAGING_DESTINATION_NAME,
                                 producerDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary)),
                 span ->
@@ -297,6 +299,7 @@ void shouldCaptureMessageHeaders(DestinationFactory destinationFactory, boolean
                             equalTo(
                                 SemanticAttributes.MESSAGING_DESTINATION_NAME,
                                 producerDestinationName),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             equalTo(SemanticAttributes.MESSAGING_MESSAGE_ID, messageId),
                             messagingTempDestination(isTemporary),
                             equalTo(

diff --git a/...ng/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaClientBaseTest.java b/...ng/src/main/java/io/opentelemetry/instrumentation/kafka/internal/KafkaClientBaseTest.java
@@ -159,6 +159,7 @@ protected static List<AttributeAssertion> sendAttributes(
             Arrays.asList(
                 equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                 equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
+                equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                 satisfies(
                     SemanticAttributes.MESSAGING_CLIENT_ID,
                     stringAssert -> stringAssert.startsWith("producer")),

diff --git a/...opentelemetry/instrumentation/kafkaclients/v2_6/InterceptorsSuppressReceiveSpansTest.java b/...opentelemetry/instrumentation/kafkaclients/v2_6/InterceptorsSuppressReceiveSpansTest.java
@@ -28,6 +28,7 @@ void assertTraces() {
                         .hasAttributesSatisfyingExactly(
                             equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                             equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
+                            equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                             satisfies(
                                 SemanticAttributes.MESSAGING_CLIENT_ID,
                                 stringAssert -> stringAssert.startsWith("producer"))),

diff --git a/...ry/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/InterceptorsTest.java b/...ry/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/InterceptorsTest.java
@@ -35,6 +35,7 @@ void assertTraces() {
                       .hasAttributesSatisfyingExactly(
                           equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                           equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
+                          equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                           satisfies(
                               SemanticAttributes.MESSAGING_CLIENT_ID,
                               stringAssert -> stringAssert.startsWith("producer"))));

diff --git a/...a/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperSuppressReceiveSpansTest.java b/...a/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperSuppressReceiveSpansTest.java
@@ -58,6 +58,7 @@ protected static List<AttributeAssertion> sendAttributes(boolean testHeaders) {
             Arrays.asList(
                 equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                 equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
+                equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                 satisfies(
                     SemanticAttributes.MESSAGING_CLIENT_ID,
                     stringAssert -> stringAssert.startsWith("producer")),

diff --git a/...library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperTest.java b/...library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/WrapperTest.java
@@ -80,6 +80,7 @@ protected static List<AttributeAssertion> sendAttributes(boolean testHeaders) {
             Arrays.asList(
                 equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                 equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
+                equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                 satisfies(
                     SemanticAttributes.MESSAGING_CLIENT_ID,
                     stringAssert -> stringAssert.startsWith("producer")),

diff --git a/...ntation/kafka/kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsDefaultTest.groovy b/...ntation/kafka/kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsDefaultTest.groovy
@@ -99,6 +99,7 @@ class KafkaStreamsDefaultTest extends KafkaStreamsBaseTest {
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "kafka"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" STREAM_PENDING
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.MESSAGING_CLIENT_ID" { it.startsWith("producer") }
             "$SemanticAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION" { it >= 0 }
             "$SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET" 0
@@ -154,6 +155,7 @@ class KafkaStreamsDefaultTest extends KafkaStreamsBaseTest {
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "kafka"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" STREAM_PROCESSED
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.MESSAGING_CLIENT_ID" { it.endsWith("producer") }
             "$SemanticAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION" { it >= 0 }
             "$SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET" 0

diff --git a/.../kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsSuppressReceiveSpansTest.groovy b/.../kafka-streams-0.11/javaagent/src/test/groovy/KafkaStreamsSuppressReceiveSpansTest.groovy
@@ -94,6 +94,7 @@ class KafkaStreamsSuppressReceiveSpansTest extends KafkaStreamsBaseTest {
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "kafka"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" STREAM_PENDING
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.MESSAGING_CLIENT_ID" "producer-1"
             "$SemanticAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION" { it >= 0 }
             "$SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET" 0
@@ -132,6 +133,7 @@ class KafkaStreamsSuppressReceiveSpansTest extends KafkaStreamsBaseTest {
           attributes {
             "$SemanticAttributes.MESSAGING_SYSTEM" "kafka"
             "$SemanticAttributes.MESSAGING_DESTINATION_NAME" STREAM_PROCESSED
+            "$SemanticAttributes.MESSAGING_OPERATION" "publish"
             "$SemanticAttributes.MESSAGING_CLIENT_ID" String
             "$SemanticAttributes.MESSAGING_KAFKA_DESTINATION_PARTITION" { it >= 0 }
             "$SemanticAttributes.MESSAGING_KAFKA_MESSAGE_OFFSET" 0

diff --git a/...est/groovy/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/PulsarClientTest.groovy b/...est/groovy/io/opentelemetry/javaagent/instrumentation/pulsar/v2_8/PulsarClientTest.groovy
@@ -587,6 +587,7 @@ class PulsarClientTest extends AgentInstrumentationSpecification {
         "$SemanticAttributes.NET_PEER_NAME" brokerHost
         "$SemanticAttributes.NET_PEER_PORT" brokerPort
         "$SemanticAttributes.MESSAGING_DESTINATION_NAME" destination
+        "$SemanticAttributes.MESSAGING_OPERATION" "publish"
         if (msgId == String) {
           "$SemanticAttributes.MESSAGING_MESSAGE_ID" String
         } else if (msgId != null) {

diff --git a/...ava/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitChannelInstrumentation.java b/...ava/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitChannelInstrumentation.java
@@ -108,11 +108,11 @@ public static void onEnter(
       Context parentContext = Java8BytecodeBridge.currentContext();
       request = ChannelAndMethod.create(channel, method);
 
-      if (!channelInstrumenter().shouldStart(parentContext, request)) {
+      if (!channelInstrumenter(request).shouldStart(parentContext, request)) {
         return;
       }
 
-      context = channelInstrumenter().start(parentContext, request);
+      context = channelInstrumenter(request).start(parentContext, request);
       CURRENT_RABBIT_CONTEXT.set(context);
       helper().setChannelAndMethod(context, request);
       scope = context.makeCurrent();
@@ -132,7 +132,7 @@ public static void stopSpan(
       scope.close();
 
       CURRENT_RABBIT_CONTEXT.remove();
-      channelInstrumenter().end(context, request, null, throwable);
+      channelInstrumenter(request).end(context, request, null, throwable);
     }
   }
 

diff --git a/...t/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitSingletons.java b/...t/src/main/java/io/opentelemetry/javaagent/instrumentation/rabbitmq/RabbitSingletons.java
@@ -30,16 +30,21 @@ public final class RabbitSingletons {
           .getBoolean("otel.instrumentation.rabbitmq.experimental-span-attributes", false);
   private static final String instrumentationName = "io.opentelemetry.rabbitmq-2.7";
   private static final Instrumenter<ChannelAndMethod, Void> channelInstrumenter =
-      createChannelInstrumenter();
+      createChannelInstrumenter(false);
+  private static final Instrumenter<ChannelAndMethod, Void> channelPublishInstrumenter =
+      createChannelInstrumenter(true);
   private static final Instrumenter<ReceiveRequest, GetResponse> receiveInstrumenter =
       createReceiveInstrumenter();
   private static final Instrumenter<DeliveryRequest, Void> deliverInstrumenter =
       createDeliverInstrumenter();
   static final ContextKey<RabbitChannelAndMethodHolder> CHANNEL_AND_METHOD_CONTEXT_KEY =
       ContextKey.named("opentelemetry-rabbitmq-channel-and-method-context-key");
 
-  public static Instrumenter<ChannelAndMethod, Void> channelInstrumenter() {
-    return channelInstrumenter;
+  public static Instrumenter<ChannelAndMethod, Void> channelInstrumenter(
+      ChannelAndMethod channelAndMethod) {
+    return channelAndMethod.getMethod().equals("Channel.basicPublish")
+        ? channelPublishInstrumenter
+        : channelInstrumenter;
   }
 
   public static Instrumenter<ReceiveRequest, GetResponse> receiveInstrumenter() {
@@ -51,21 +56,19 @@ static Instrumenter<DeliveryRequest, Void> deliverInstrumenter() {
   }
 
   @SuppressWarnings("deprecation") // have to use the deprecated Net*AttributesExtractor for now
-  private static Instrumenter<ChannelAndMethod, Void> createChannelInstrumenter() {
+  private static Instrumenter<ChannelAndMethod, Void> createChannelInstrumenter(boolean publish) {
     return Instrumenter.<ChannelAndMethod, Void>builder(
             GlobalOpenTelemetry.get(), instrumentationName, ChannelAndMethod::getMethod)
         .addAttributesExtractor(
             buildMessagingAttributesExtractor(
-                RabbitChannelAttributesGetter.INSTANCE, MessageOperation.PUBLISH))
+                RabbitChannelAttributesGetter.INSTANCE, publish ? MessageOperation.PUBLISH : null))
         .addAttributesExtractor(
             io.opentelemetry.instrumentation.api.instrumenter.net.NetClientAttributesExtractor
                 .create(new RabbitChannelNetAttributesGetter()))
         .addContextCustomizer(
             (context, request, startAttributes) ->
                 context.with(CHANNEL_AND_METHOD_CONTEXT_KEY, new RabbitChannelAndMethodHolder()))
-        .buildInstrumenter(
-            channelAndMethod ->
-                channelAndMethod.getMethod().equals("Channel.basicPublish") ? PRODUCER : CLIENT);
+        .buildInstrumenter(channelAndMethod -> publish ? PRODUCER : CLIENT);
   }
 
   @SuppressWarnings("deprecation") // have to use the deprecated Net*AttributesExtractor for now

diff --git a/.../opentelemetry/javaagent/instrumentation/reactor/kafka/v1_0/AbstractReactorKafkaTest.java b/.../opentelemetry/javaagent/instrumentation/reactor/kafka/v1_0/AbstractReactorKafkaTest.java
@@ -180,6 +180,7 @@ protected static List<AttributeAssertion> sendAttributes(ProducerRecord<String,
             asList(
                 equalTo(SemanticAttributes.MESSAGING_SYSTEM, "kafka"),
                 equalTo(SemanticAttributes.MESSAGING_DESTINATION_NAME, record.topic()),
+                equalTo(SemanticAttributes.MESSAGING_OPERATION, "publish"),
                 satisfies(
                     SemanticAttributes.MESSAGING_CLIENT_ID,
                     stringAssert -> stringAssert.startsWith("producer")),