Stop onErrorDropped events from leaking into default Reactor Hook

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/RntbdTransportClient.java

@@ -28,8 +28,10 @@
 import io.netty.handler.ssl.SslContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Hooks;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.SignalType;
+import reactor.util.context.Context;
 
 import java.io.File;
 import java.io.IOException;
@@ -41,6 +43,7 @@
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 
 import static com.azure.cosmos.implementation.directconnectivity.rntbd.RntbdReporter.reportIssue;
 import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkArgument;
@@ -58,6 +61,26 @@ public final class RntbdTransportClient extends TransportClient {
     private static final AtomicLong instanceCount = new AtomicLong();
     private static final Logger logger = LoggerFactory.getLogger(RntbdTransportClient.class);
 
+    /**
+     * A key that can be used to store a sequence-specific {@link Hooks#onErrorDropped(Consumer)}
+     * hook in a {@link Context}, as a {@link Consumer Consumer<Throwable>}.
+     */
+    private static final String KEY_ON_ERROR_DROPPED = "reactor.onErrorDropped.local";
+
+    /**
+     * This lambda gets injected into the local Reactor Context to react tot he onErrorDropped event and
+     * log the throwable with DEBUG level instead of the ERROR level used in the default hook.
+     * This is safe here because we guarantee resource clean-up with the doFinally-lambda
+     */
+    private static final Consumer<? super Throwable> onErrorDropHookWithReduceLogLevel =
+        throwable -> {
+            if (logger.isDebugEnabled()) {
+                logger.debug(
+                    "Extra error - on error dropped - operator called :",
+                    throwable);
+            }
+        };
+
     private final AtomicBoolean closed = new AtomicBoolean();
     private final RntbdEndpoint.Provider endpointProvider;
     private final long id;
@@ -128,6 +151,8 @@ public Mono<StoreResponse> invokeStoreAsync(final Uri addressUri, final RxDocume
         final RntbdEndpoint endpoint = this.endpointProvider.get(address);
         final RntbdRequestRecord record = endpoint.request(requestArgs);
 
+        final Context reactorContext = Context.of(KEY_ON_ERROR_DROPPED, onErrorDropHookWithReduceLogLevel);
+
         final Mono<StoreResponse> result = Mono.fromFuture(record.whenComplete((response, throwable) -> {
 
             record.stage(RntbdRequestRecord.Stage.COMPLETED);
@@ -177,7 +202,10 @@ public Mono<StoreResponse> invokeStoreAsync(final Uri addressUri, final RxDocume
             // This lambda does not prevent requests that complete exceptionally before the call to this lambda from
             // bubbling up to Hooks#onErrorDropped as CompletionException errors. We will still see some onErrorDropped
             // messages due to CompletionException errors. Anecdotal evidence shows that this is more likely to be seen
-            // in low latency environments on Azure cloud.
+            // in low latency environments on Azure cloud. To avoid the onErrorDropped events to get logged in the
+            // default hook (which logs with level ERROR) we inject a local hook in the Reactor Context to just log it
+            // as DEBUG level fro the lifecycle of this Mono (safe here because we know the onErrorDropped doesn't have
+            // any functional issues.
             //
             // One might be tempted to complete a pending request here, but that is ill advised. Testing and
             // inspection of the reactor code shows that this does not prevent errors from bubbling up to
@@ -189,11 +217,6 @@ public Mono<StoreResponse> invokeStoreAsync(final Uri addressUri, final RxDocume
             //
             // * Calling record.complete with a null value causes failures in all environments, depending on the
             //   operation being performed. In short: many of our tests fail.
-            //
-            // TODO (DANOBLE) verify the correctness of this statement: Fact: We still see some of these errors. Does
-            //  reactor provide a mechanism other than Hooks#onErrorDropped(Consumer<? super Throwable> consumer) for
-            //  doing this per Mono or must we, for example, rely on something like this in the consistency layer:
-            //  https://www.codota.com/code/java/classes/reactor.core.publisher.Hooks
 
             if (signalType != SignalType.CANCEL) {
                 return;
@@ -220,7 +243,7 @@ public Mono<StoreResponse> invokeStoreAsync(final Uri addressUri, final RxDocume
                             RntbdObjectMapper.toJson(throwable));
                     }
                 });
-        });
+        }).subscriberContext(reactorContext);
     }
 
     public Tag tag() {
@@ -408,7 +431,7 @@ public int maxRequestsPerChannel() {
         public int maxConcurrentRequestsPerEndpoint() {
             if (this.maxConcurrentRequestsPerEndpointOverride > 0) {
                 return maxConcurrentRequestsPerEndpointOverride;
-            };
+            }
 
             return Math.max(
                 DEFAULT_MIN_MAX_CONCURRENT_REQUESTS_PER_ENDPOINT,
@@ -419,10 +442,6 @@ public Duration receiveHangDetectionTime() {
             return this.receiveHangDetectionTime;
         }
 
-        public Duration requestExpiryInterval() {
-            return this.requestExpiryInterval;
-        }
-
         public Duration requestTimeout() {
             return this.requestTimeout;
         }
@@ -570,7 +589,7 @@ public static class Builder {
             }
 
             private int bufferPageSize;
-            private Duration connectionAcquisitionTimeout;
+            private final Duration connectionAcquisitionTimeout;
             private Duration connectTimeout;
             private Duration idleChannelTimeout;
             private Duration idleChannelTimerResolution;

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/directconnectivity/rntbd/RntbdClientChannelPool.java

@@ -1470,33 +1470,27 @@ public final void run() {
             long currentNanoTime = System.nanoTime();
 
             while (true) {
-                try {
-                    AcquireListener removedTask = this.pool.pendingAcquisitions.poll();
-                    if (removedTask == null) {
-                        // queue is empty
-                        break;
-                    }
+                AcquireListener removedTask = this.pool.pendingAcquisitions.poll();
+                if (removedTask == null) {
+                    // queue is empty
+                    break;
+                }
 
-                    long expiryTime = removedTask.getAcquisitionTimeoutInNanos();
+                long expiryTime = removedTask.getAcquisitionTimeoutInNanos();
 
-                    // Compare nanoTime as described in the System.nanoTime documentation
-                    // See:
-                    // * https://docs.oracle.com/javase/7/docs/api/java/lang/System.html#nanoTime()
-                    // * https://github.com/netty/netty/issues/3705
-                    if (expiryTime - currentNanoTime < 0) {
-                        this.onTimeout(removedTask);
-                    } else {
-                        if (!this.pool.pendingAcquisitions.offer(removedTask)) {
-                            logger.error("Unexpected failure when returning the removed task"
-                                    + " to pending acquisition queue. current size [{}]",
-                                this.pool.pendingAcquisitions.size());
-                        }
-                        break;
+                // Compare nanoTime as described in the System.nanoTime documentation
+                // See:
+                // * https://docs.oracle.com/javase/7/docs/api/java/lang/System.html#nanoTime()
+                // * https://github.com/netty/netty/issues/3705
+                if (expiryTime - currentNanoTime <= 0) {
+                    this.onTimeout(removedTask);
+                } else {
+                    if (!this.pool.pendingAcquisitions.offer(removedTask)) {
+                        logger.error("Unexpected failure when returning the removed task"
+                                + " to pending acquisition queue. current size [{}]",
+                            this.pool.pendingAcquisitions.size());
                     }
-                } catch (Exception e) {
-                    logger.error("Unexpected failure in clearing the expired tasks"
-                        + " in pending acquisition queue. current size [{}]",
-                        this.pool.pendingAcquisitions.size(), e);
+                    break;
                 }
             }
         }