Add id to log AmqpChannelProcessor.ChannelSubscriber lifetime (Azure#29738)

* Add id to log AmqpChannelProcessor.ChannelSubscriber lifetime

* add subscriber id to reactor subscriber context

Author: Liudmila Molkova

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/AmqpChannelProcessor.java

@@ -30,6 +30,7 @@
 
 import static com.azure.core.amqp.implementation.ClientConstants.ENTITY_PATH_KEY;
 import static com.azure.core.amqp.implementation.ClientConstants.INTERVAL_KEY;
+import static com.azure.core.amqp.implementation.ClientConstants.SUBSCRIBER_ID_KEY;
 
 public class AmqpChannelProcessor<T> extends Mono<T> implements Processor<T, T>, CoreSubscriber<T>, Disposable {
     @SuppressWarnings("rawtypes")
@@ -66,7 +67,6 @@ public AmqpChannelProcessor(String fullyQualifiedNamespace, String entityPath,
         this.endpointStatesFunction = Objects.requireNonNull(endpointStatesFunction,
             "'endpointStates' cannot be null.");
         this.retryPolicy = Objects.requireNonNull(retryPolicy, "'retryPolicy' cannot be null.");
-
         Map<String, Object> loggingContext = new HashMap<>(1);
         loggingContext.put(ENTITY_PATH_KEY, Objects.requireNonNull(entityPath, "'entityPath' cannot be null."));
         this.logger = new ClientLogger(getClass(), loggingContext);
@@ -77,6 +77,7 @@ public AmqpChannelProcessor(String fullyQualifiedNamespace, Function<T, Flux<Amq
         this.endpointStatesFunction = Objects.requireNonNull(endpointStatesFunction,
             "'endpointStates' cannot be null.");
         this.retryPolicy = Objects.requireNonNull(retryPolicy, "'retryPolicy' cannot be null.");
+
         this.logger = new ClientLogger(getClass(), Objects.requireNonNull(loggingContext, "'loggingContext' cannot be null."));
         this.errorContext = new AmqpErrorContext(fullyQualifiedNamespace);
     }
@@ -107,8 +108,6 @@ public void onNext(T amqpChannel) {
             currentChannel = amqpChannel;
 
             final ConcurrentLinkedDeque<ChannelSubscriber<T>> currentSubscribers = subscribers;
-            logger.info("Next AMQP channel received, updating {} current subscribers", subscribers.size());
-
             currentSubscribers.forEach(subscription -> subscription.onNext(amqpChannel));
 
             connectionSubscription = endpointStatesFunction.apply(amqpChannel).subscribe(
@@ -227,8 +226,6 @@ public void onError(Throwable throwable) {
             synchronized (lock) {
                 final ConcurrentLinkedDeque<ChannelSubscriber<T>> currentSubscribers = subscribers;
                 subscribers = new ConcurrentLinkedDeque<>();
-                logger.info("Error in AMQP channel processor. Notifying {} subscribers.", currentSubscribers.size());
-
                 currentSubscribers.forEach(subscriber -> subscriber.onError(throwable));
             }
         }
@@ -242,7 +239,6 @@ public void onComplete() {
         synchronized (lock) {
             final ConcurrentLinkedDeque<ChannelSubscriber<T>> currentSubscribers = subscribers;
             subscribers = new ConcurrentLinkedDeque<>();
-            logger.info("AMQP channel processor completed. Notifying {} subscribers.", currentSubscribers.size());
             currentSubscribers.forEach(subscriber -> subscriber.onComplete());
         }
     }
@@ -270,8 +266,8 @@ public void subscribe(CoreSubscriber<? super T> actual) {
             }
         }
 
+        subscriber.onAdd();
         subscribers.add(subscriber);
-        logger.atVerbose().addKeyValue("subscribers", subscribers.size()).log("Added a subscriber.");
 
         if (!isRetryPending.get()) {
             requestUpstream();
@@ -370,17 +366,37 @@ private void close(T channel) {
      * which removes itself from the tracking list, then propagates the notification to the wrapped subscriber.
      */
     private static final class ChannelSubscriber<T> extends Operators.MonoSubscriber<T, T> {
+
         private final AmqpChannelProcessor<T> processor;
+        // subscriberId is only needed for logging and not every subscriber is logged, so let's make it lazy.
+        private String subscriberId = null;
 
         private ChannelSubscriber(CoreSubscriber<? super T> actual, AmqpChannelProcessor<T> processor) {
             super(actual);
             this.processor = processor;
         }
 
+        void onAdd() {
+            Object subscriberIdObj = actual.currentContext().getOrDefault(SUBSCRIBER_ID_KEY, null);
+            if (subscriberIdObj != null) {
+                subscriberId = subscriberIdObj.toString();
+            } else {
+                subscriberId = StringUtil.getRandomString("un");
+            }
+
+            // most subscribers never get here and will be completed immediately after they are created.
+            processor.logger.atVerbose()
+                .addKeyValue(SUBSCRIBER_ID_KEY, subscriberId)
+                .log("Added subscriber.");
+        }
+
         @Override
         public void cancel() {
             processor.subscribers.remove(this);
             super.cancel();
+            processor.logger.atVerbose()
+                .addKeyValue(SUBSCRIBER_ID_KEY, subscriberId)
+                .log("Canceled subscriber");
         }
 
         @Override
@@ -389,6 +405,9 @@ public void onComplete() {
                 // first untrack before calling into external code.
                 processor.subscribers.remove(this);
                 actual.onComplete();
+                processor.logger.atInfo()
+                    .addKeyValue(SUBSCRIBER_ID_KEY, subscriberId)
+                    .log("AMQP channel processor completed.");
             }
         }
 
@@ -397,6 +416,10 @@ public void onNext(T channel) {
             if (!isCancelled()) {
                 processor.subscribers.remove(this);
                 super.complete(channel);
+
+                processor.logger.atInfo()
+                    .addKeyValue(SUBSCRIBER_ID_KEY, subscriberId)
+                    .log("Next AMQP channel received.");
             }
         }
 
@@ -405,6 +428,9 @@ public void onError(Throwable throwable) {
             if (!isCancelled()) {
                 processor.subscribers.remove(this);
                 actual.onError(throwable);
+                processor.logger.atInfo()
+                    .addKeyValue(SUBSCRIBER_ID_KEY, subscriberId)
+                    .log("Error in AMQP channel processor.");
             } else {
                 Operators.onErrorDropped(throwable, currentContext());
             }

sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ClientConstants.java

@@ -24,6 +24,7 @@ public final class ClientConstants {
     public static final String SIGNAL_TYPE_KEY = "signalType";
     public static final String HOSTNAME_KEY = "hostName";
     public static final String INTERVAL_KEY = "interval_ms";
+    public static final String SUBSCRIBER_ID_KEY = "subscriberId";
 
     /**
      * The default maximum allowable size, in bytes, for a batch to be sent.