Address review comments from copilot-pull-request-reviewer: fix stable batch receive destination

Author: trask

instrumentation/rocketmq/rocketmq-client-4.8/library/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/RocketMqInstrumenterFactory.java

@@ -38,6 +38,8 @@ class RocketMqInstrumenterFactory {
   // copied from MessagingIncubatingAttributes
   private static final AttributeKey<String> MESSAGING_CONSUMER_GROUP_NAME =
       AttributeKey.stringKey("messaging.consumer.group.name");
+  private static final AttributeKey<String> MESSAGING_DESTINATION_NAME =
+      AttributeKey.stringKey("messaging.destination.name");
   private static final AttributeKey<String> MESSAGING_OPERATION =
       AttributeKey.stringKey("messaging.operation");
   private static final AttributeKey<String> MESSAGING_OPERATION_NAME =
@@ -119,6 +121,7 @@ public void onStart(
                     RocketMqConsumerRequest request) {
                   attributes.put(MESSAGING_BATCH_MESSAGE_COUNT, request.getBatchSize());
                   attributes.put(MESSAGING_CONSUMER_GROUP_NAME, request.getConsumerGroup());
+                  attributes.put(MESSAGING_DESTINATION_NAME, request.getMessage().getTopic());
                   attributes.put(MESSAGING_ROCKETMQ_NAMESPACE, request.getNamespace());
                 }
 
@@ -209,8 +212,10 @@ private static <T, R> AttributesExtractor<T, R> buildMessagingAttributesExtracto
         .build();
   }
 
-  private static String spanNameOnReceive(RocketMqConsumerRequest unused) {
-    return emitStableMessagingSemconv() ? "receive multiple_sources" : "multiple_sources receive";
+  private static String spanNameOnReceive(RocketMqConsumerRequest request) {
+    return emitStableMessagingSemconv()
+        ? "receive " + request.getMessage().getTopic()
+        : "multiple_sources receive";
   }
 
   private RocketMqInstrumenterFactory() {}

instrumentation/rocketmq/rocketmq-client-4.8/testing/src/main/java/io/opentelemetry/instrumentation/rocketmqclient/v4_8/AbstractRocketMqClientTest.java

@@ -422,13 +422,16 @@ private void runBatchConsumeTest(boolean failConsumption) throws Exception {
                   span ->
                       span.hasName(
                               emitStableMessagingSemconv()
-                                  ? "receive multiple_sources"
+                                  ? "receive " + sharedTopic
                                   : "multiple_sources receive")
                           .hasKind(emitStableMessagingSemconv() ? CLIENT : CONSUMER)
                           .hasAttributesSatisfyingExactly(
                               equalTo(MESSAGING_SYSTEM, "rocketmq"),
                               namespace(),
                               consumerGroup(),
+                              equalTo(
+                                  MESSAGING_DESTINATION_NAME,
+                                  emitStableMessagingSemconv() ? sharedTopic : null),
                               equalTo(
                                   MESSAGING_BATCH_MESSAGE_COUNT,
                                   emitStableMessagingSemconv() ? Long.valueOf(2) : null),