Address spring-messaging-azure-service api review comments

sdk/spring/spring-cloud-azure-stream-binder-servicebus/src/main/java/com/azure/spring/cloud/stream/binder/servicebus/ServiceBusMessageChannelBinder.java

@@ -50,6 +50,7 @@
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
+import org.springframework.util.StringUtils;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -287,8 +288,11 @@ private ServiceBusProcessorFactory getProcessorFactory() {
 
             processorFactoryCustomizers.forEach(customizer -> customizer.customize(this.processorFactory));
 
-            this.processorFactory.addListener((name, subscription, client) -> {
-                String instrumentationName = name + "/" + getGroup(subscription);
+            this.processorFactory.addListener((name, client) -> {
+                String subscriptionName = client.getSubscriptionName();
+                boolean isTopic = StringUtils.hasText(subscriptionName);
+                String entityName = isTopic ? client.getTopicName() : client.getQueueName();
+                String instrumentationName = entityName + "/" + getGroup(subscriptionName);
                 Instrumentation instrumentation = new ServiceBusProcessorInstrumentation(instrumentationName, CONSUMER, Duration.ofMinutes(2));
                 instrumentation.setStatus(Instrumentation.Status.UP);
                 instrumentationManager.addHealthInstrumentation(instrumentation);

sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusProcessorFactory.java

@@ -97,19 +97,17 @@ interface Listener {
          * The callback method that the processor has been added.
          *
          * @param name the name for the processor.
-         * @param subscription the subscription for the processor.
          * @param client the client for the processor.
          */
-        void processorAdded(String name, String subscription, ServiceBusProcessorClient client);
+        void processorAdded(String name, ServiceBusProcessorClient client);
 
         /**
          * The default callback method that the processor has been removed.
          *
          * @param name the name for the processor.
-         * @param subscription the subscription for the processor.
          * @param client the client for the processor.
          */
-        default void processorRemoved(String name, String subscription, ServiceBusProcessorClient client) {
+        default void processorRemoved(String name, ServiceBusProcessorClient client) {
         }
 
     }

sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/implementation/core/DefaultServiceBusNamespaceProcessorFactory.java

@@ -6,9 +6,9 @@
 import com.azure.core.credential.TokenCredential;
 import com.azure.messaging.servicebus.ServiceBusClientBuilder;
 import com.azure.messaging.servicebus.ServiceBusProcessorClient;
-import com.azure.spring.cloud.core.implementation.util.AzureSpringIdentifier;
 import com.azure.spring.cloud.core.credential.AzureCredentialResolver;
 import com.azure.spring.cloud.core.customizer.AzureServiceClientBuilderCustomizer;
+import com.azure.spring.cloud.core.implementation.util.AzureSpringIdentifier;
 import com.azure.spring.cloud.service.implementation.servicebus.factory.ServiceBusProcessorClientBuilderFactory;
 import com.azure.spring.cloud.service.implementation.servicebus.factory.ServiceBusSessionProcessorClientBuilderFactory;
 import com.azure.spring.cloud.service.listener.MessageListener;
@@ -199,7 +199,7 @@ private ServiceBusProcessorClient doCreateProcessor(String name,
                 client = builder.buildProcessorClient();
             }
 
-            this.listeners.forEach(l -> l.processorAdded(k.getDestination(), k.getGroup(), client));
+            this.listeners.forEach(l -> l.processorAdded(buildProcessorName(k), client));
             return client;
         });
     }
@@ -228,7 +228,7 @@ public void setDefaultAzureCredential(TokenCredential defaultAzureCredential) {
     private void close(Map<ConsumerIdentifier, ServiceBusProcessorClient> map, Consumer<ServiceBusProcessorClient> close) {
         map.forEach((t, p) -> {
             try {
-                listeners.forEach(l -> l.processorRemoved(t.getDestination(), t.getGroup(), p));
+                listeners.forEach(l -> l.processorRemoved(buildProcessorName(t), p));
                 close.accept(p);
             } catch (Exception ex) {
                 LOGGER.warn("Failed to clean service bus queue client factory", ex);
@@ -288,6 +288,11 @@ private void customizeBuilder(String entityName, String subscription,
                                  .forEach(customizer -> customizer.getSessionCustomizer().customize(builder));
     }
 
+    private String buildProcessorName(ConsumerIdentifier k) {
+        String group = k.getGroup();
+        return k.getDestination() + "/" + (group == null ? "" : group);
+    }
+
     public static class ServiceBusProcessClientBuilderCustomizer {
 
         private final AzureServiceClientBuilderCustomizer<ServiceBusClientBuilder.ServiceBusProcessorClientBuilder> noneSessionCustomizer;

sdk/spring/spring-messaging-azure-servicebus/src/test/java/com/azure/spring/messaging/servicebus/implementation/core/DefaultServiceBusNamespaceProcessorFactoryTests.java

@@ -34,8 +34,8 @@ void setUp() {
         this.processorFactory = new DefaultServiceBusNamespaceProcessorFactory(namespaceProperties);
         queueProcessorAddedTimes = 0;
         topicProcessorAddedTimes = 0;
-        this.processorFactory.addListener((name, subscription, client) -> {
-            if (subscription == null) {
+        this.processorFactory.addListener((name, client) -> {
+            if (client.getSubscriptionName() == null) {
                 queueProcessorAddedTimes++;
             } else {
                 topicProcessorAddedTimes++;