review

Azure · Apr 25, 2024 · cf49232 · cf49232
1 parent 7cb225e
commit cf49232
Show file tree

Hide file tree

Showing 4 changed files with 13 additions and 16 deletions.
diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/MessageFlux.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/MessageFlux.java
@@ -983,6 +983,9 @@ private Mono<Void> updateDisposition(String deliveryTag, DeliveryState deliveryS
                 // RecoverableReactorReceiver, Or downstream cancels parent RecoverableReactorReceiver.
                 // 1 & 2 means we don't have to read the 'volatile' variables 'parent.done', 'parent.cancelled'
                 //
+                final String state
+                    = String.format("[link.done:%b link.cancelled:%b parent.done:%b parent.cancelled:%b]", done,
+                        s == CANCELLED_SUBSCRIPTION, parent.done, parent.cancelled);
 
                 final DeliveryNotOnLinkException dispositionError
                     = DeliveryNotOnLinkException.linkClosed(deliveryTag, deliveryState);
@@ -994,16 +997,10 @@ private Mono<Void> updateDisposition(String deliveryTag, DeliveryState deliveryS
                 if (upstreamError != null) {
                     dispositionError.addSuppressed(upstreamError);
                 }
-
-                LoggingEventBuilder log = logger.atError()
+                return monoError(logger.atError()
                     .addKeyValue(DELIVERY_TAG_KEY, deliveryTag)
                     .addKeyValue(DELIVERY_STATE_KEY, deliveryState)
-                    .addKeyValue("link.done", done)
-                    .addKeyValue("link.cancelled", s == CANCELLED_SUBSCRIPTION)
-                    .addKeyValue("parent.done", parent.done)
-                    .addKeyValue("parent.cancelled", parent.cancelled);
-
-                return monoError(log, dispositionError);
+                    .addKeyValue("messageFluxState", state), dispositionError);
             }
             return receiver.updateDisposition(deliveryTag, deliveryState);
         }

diff --git a/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorDispatcher.java b/...e/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorDispatcher.java
@@ -247,7 +247,7 @@ public void run(Selectable selectable) {
                     ioSignal.sink().close();
                 }
             } catch (IOException ioException) {
-                logger.warning("CloseHandler.sink().close() failed with an error.", ioException);
+                logger.error("CloseHandler.sink().close() failed with an error.", ioException);
             }
 
             workScheduler.run(null);
@@ -257,7 +257,7 @@ public void run(Selectable selectable) {
                     ioSignal.source().close();
                 }
             } catch (IOException ioException) {
-                logger.warning("CloseHandler.source().close() failed with an error.", ioException);
+                logger.error("CloseHandler.source().close() failed with an error.", ioException);
             }
         }
     }

diff --git a/...core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java b/...core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/ReactorSession.java
@@ -358,8 +358,8 @@ protected Mono<AmqpReceiveLink> createConsumer(String linkName, String entityPat
                 .addKeyValue(ENTITY_PATH_KEY, entityPath)
                 .addKeyValue(LINK_NAME_KEY, linkName);
 
-            return monoError(logBuilder, Exceptions.propagate(new AmqpException(true,
-                "Cannot create receive link from a closed session.", sessionHandler.getErrorContext())));
+            return monoError(logBuilder, new AmqpException(true,
+                "Cannot create receive link from a closed session.", sessionHandler.getErrorContext()));
         }
 
         final LinkSubscription<AmqpReceiveLink> existingLink = openReceiveLinks.get(linkName);
@@ -448,8 +448,8 @@ private Mono<AmqpSendLink> createProducer(String linkName, String entityPath,
                 .addKeyValue(ENTITY_PATH_KEY, entityPath)
                 .addKeyValue(LINK_NAME_KEY, linkName);
 
-            return monoError(logBuilder, Exceptions.propagate(new AmqpException(true,
-                "Cannot create send link from a closed session.", sessionHandler.getErrorContext())));
+            return monoError(logBuilder, new AmqpException(true,
+                "Cannot create send link from a closed session.", sessionHandler.getErrorContext()));
         }
 
         final LinkSubscription<AmqpSendLink> existing = openSendLinks.get(linkName);

diff --git a/...icebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java b/...icebus/src/main/java/com/azure/messaging/servicebus/implementation/ManagementChannel.java
@@ -524,9 +524,9 @@ public Flux<RuleProperties> listRules() {
             } else if (statusCode == AmqpResponseCode.NO_CONTENT) {
                 list = Collections.emptyList();
             } else {
-                throw logger.logExceptionAsWarning(Exceptions.propagate(new AmqpException(true,
+                throw logger.logExceptionAsWarning(new AmqpException(true,
                     "Get rules response error. Could not get rules.",
-                    getErrorContext())));
+                    getErrorContext()));
             }
 
             return Flux.fromIterable(list);