Add exception handler on HTTP/2 parent channel to suppress WARN logs

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/http/Http2ParentChannelExceptionHandlerTest.java

@@ -0,0 +1,227 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.implementation.http;
+
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.http2.Http2FrameCodec;
+import io.netty.handler.codec.http2.Http2FrameCodecBuilder;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
+import org.testng.annotations.Test;
+
+import java.io.IOException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Verifies that {@link Http2ParentChannelExceptionHandler} uses connection
+ * state — active stream count and channel activity — to determine whether
+ * exceptions are logged at DEBUG (suppressed) or WARN (preserved).
+ * Exception type is NOT a filtering dimension.
+ *
+ * The EmbeddedChannel is configured to mirror the production HTTP/2 parent
+ * channel pipeline:
+ * <pre>
+ *   Http2FrameCodec → Http2MultiplexHandler → Http2ParentChannelExceptionHandler → TailContext
+ * </pre>
+ * (SslHandler is omitted because it requires an SSLContext and is not relevant
+ * to exception propagation behavior.)
+ *
+ * {@code checkException()} re-throws any exception that reached the pipeline tail.
+ */
+public class Http2ParentChannelExceptionHandlerTest {
+
+    /**
+     * BEFORE fix — without the handler, exceptions reach the pipeline tail.
+     * EmbeddedChannel's checkException() re-throws the unhandled exception,
+     * proving it reached Netty's TailContext (which in production logs as WARN).
+     */
+    @Test(groups = "unit")
+    public void withoutHandler_exceptionReachesTail() {
+        EmbeddedChannel channel = createH2ParentChannel(false);
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("Connection reset by peer"));
+
+        assertThatThrownBy(channel::checkException)
+            .isInstanceOf(IOException.class)
+            .hasMessageContaining("Connection reset by peer");
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * With handler — exception on idle connection (0 active streams) is
+     * consumed at DEBUG. The suppression is based on connection state
+     * (no active streams), not exception type.
+     *
+     * In production, channel.isActive() transitions to false during the
+     * RST handling cycle, satisfying the OR condition. In EmbeddedChannel
+     * we can only verify the activeStreams == 0 branch.
+     */
+    @Test(groups = "unit")
+    public void withHandler_zeroActiveStreams_consumedAtDebug() {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        Http2FrameCodec codec = channel.pipeline().get(Http2FrameCodec.class);
+        assertThat(codec).isNotNull();
+        assertThat(codec.connection().numActiveStreams()).isEqualTo(0);
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("recvAddress(..) failed with error(-104): Connection reset by peer"));
+
+        // Exception consumed — does NOT reach tail
+        channel.checkException();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * Handler does not close the channel — connection lifecycle is managed
+     * by reactor-netty's pool eviction, not by this handler.
+     */
+    @Test(groups = "unit")
+    public void withHandler_exceptionDoesNotCloseChannel() {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        assertThat(channel.isActive()).isTrue();
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("Connection reset by peer"));
+
+        channel.checkException();
+        assertThat(channel.isOpen()).isTrue();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * RuntimeException on idle connection is also consumed — suppression
+     * is based on connection state, not exception type.
+     */
+    @Test(groups = "unit")
+    public void withHandler_runtimeException_zeroActiveStreams_consumed() {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        channel.pipeline().fireExceptionCaught(
+            new RuntimeException("Unexpected state error"));
+
+        channel.checkException();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * NullPointerException on idle connection is also consumed — same
+     * connection-state-based suppression regardless of exception type.
+     */
+    @Test(groups = "unit")
+    public void withHandler_npe_zeroActiveStreams_consumed() {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        channel.pipeline().fireExceptionCaught(
+            new NullPointerException("handler bug"));
+
+        channel.checkException();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * With handler — exception on a connection with active streams is
+     * consumed (does not reach TailContext). The handler logs at WARN
+     * instead of DEBUG because in-flight requests may be affected.
+     */
+    @Test(groups = "unit")
+    public void withHandler_activeStreams_consumedAtWarn() throws Exception {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        Http2FrameCodec codec = channel.pipeline().get(Http2FrameCodec.class);
+        assertThat(codec).isNotNull();
+
+        // Create an active stream (client-initiated, odd stream ID)
+        codec.connection().local().createStream(1, false);
+        assertThat(codec.connection().numActiveStreams()).isEqualTo(1);
+        assertThat(channel.isActive()).isTrue();
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("Connection reset by peer"));
+
+        // Exception consumed — does NOT reach tail, even with active streams
+        channel.checkException();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * Handler does not close the channel even when active streams exist —
+     * connection lifecycle is managed by reactor-netty's pool eviction.
+     */
+    @Test(groups = "unit")
+    public void withHandler_activeStreams_channelNotClosed() throws Exception {
+        EmbeddedChannel channel = createH2ParentChannel(true);
+
+        Http2FrameCodec codec = channel.pipeline().get(Http2FrameCodec.class);
+        assertThat(codec).isNotNull();
+
+        codec.connection().local().createStream(1, false);
+        assertThat(codec.connection().numActiveStreams()).isEqualTo(1);
+        assertThat(channel.isActive()).isTrue();
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("Connection reset by peer"));
+
+        channel.checkException();
+        assertThat(channel.isOpen()).isTrue();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * With handler — when Http2FrameCodec is absent from the pipeline,
+     * getActiveStreamCount() returns -1. Since -1 != 0 and the channel
+     * is active, the handler takes the safe WARN path. This covers the
+     * fallback behavior when the codec is unavailable (e.g., torn down
+     * during shutdown).
+     */
+    @Test(groups = "unit")
+    public void withHandler_codecAbsent_fallsBackToWarnPath() {
+        EmbeddedChannel channel = new EmbeddedChannel(
+            new Http2ParentChannelExceptionHandler());
+
+        assertThat(channel.pipeline().get(Http2FrameCodec.class)).isNull();
+        assertThat(channel.isActive()).isTrue();
+
+        channel.pipeline().fireExceptionCaught(
+            new IOException("Connection reset by peer"));
+
+        // Exception consumed — does NOT reach tail
+        channel.checkException();
+        assertThat(channel.isOpen()).isTrue();
+
+        channel.finishAndReleaseAll();
+    }
+
+    /**
+     * Creates an EmbeddedChannel matching the production HTTP/2 parent channel
+     * pipeline (minus SslHandler): Http2FrameCodec → Http2MultiplexHandler →
+     * Http2ParentChannelExceptionHandler.
+     */
+    private static EmbeddedChannel createH2ParentChannel(boolean withExceptionHandler) {
+        Http2FrameCodec codec = Http2FrameCodecBuilder.forClient()
+            .autoAckSettingsFrame(true)
+            .build();
+
+        Http2MultiplexHandler multiplexHandler = new Http2MultiplexHandler(
+            new ChannelInboundHandlerAdapter());
+
+        if (withExceptionHandler) {
+            return new EmbeddedChannel(codec, multiplexHandler,
+                new Http2ParentChannelExceptionHandler());
+        } else {
+            return new EmbeddedChannel(codec, multiplexHandler);
+        }
+    }
+}

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -13,6 +13,7 @@
 * Fixed an issue where `CustomItemSerializer` was incorrectly applied to internal SDK query pipeline structures (e.g., `OrderByRowResult`, `Document`), causing deserialization failures in ORDER BY, GROUP BY, aggregate, DISTINCT, and hybrid search queries. - See [PR 48811](https://github.com/Azure/azure-sdk-for-java/pull/48811)
 * Fixed an issue where `SqlParameter` ignored the configured `CustomItemSerializer`, always using the internal default serializer instead. - See [PR 48811](https://github.com/Azure/azure-sdk-for-java/pull/48811)
 * Fixed a `ClientTelemetry` static initialization failure when IMDS access is disabled, preventing `NoClassDefFoundError` during Cosmos client creation in non-Azure environments. - See [PR 48888](https://github.com/Azure/azure-sdk-for-java/pull/48888)
+* Fixed an issue where Netty could log "An exceptionCaught() event was fired, and it reached at the tail of the pipeline" on HTTP/2 connections when the server resets idle TCP connections by adding an exception handler on the HTTP/2 parent channel to handle these connection-level exceptions more appropriately. - See [PR 48890](https://github.com/Azure/azure-sdk-for-java/pull/48890)
 
 #### Other Changes
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2ParentChannelExceptionHandler.java

@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.implementation.http;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http2.Http2FrameCodec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Exception handler for the HTTP/2 parent (TCP) channel pipeline.
+ * <p>
+ * In HTTP/2, reactor-netty multiplexes streams on a shared parent TCP connection.
+ * Child stream channels have {@code ChannelOperationsHandler} which catches exceptions
+ * and fails the active subscriber (matching HTTP/1.1 behavior). However, the parent
+ * channel has no such handler — exceptions propagate to Netty's {@code TailContext}
+ * which logs them as WARN ("An exceptionCaught() event was fired, and it reached at
+ * the tail of the pipeline").
+ * <p>
+ * This handler consumes all exceptions on the parent channel and uses connection
+ * state to decide the log level:
+ * <ul>
+ *   <li><b>DEBUG</b> — when {@code activeStreams == 0} OR {@code !channelActive}.
+ *       No in-flight requests are affected (e.g., TCP RST from LB idle timeout,
+ *       post-close cleanup).</li>
+ *   <li><b>WARN</b> — when active streams exist on a live channel. The exception
+ *       may affect in-flight requests and is worth alerting on.</li>
+ * </ul>
+ * <p>
+ * The handler does NOT close the channel or alter connection lifecycle — reactor-netty
+ * and the connection pool's eviction predicate ({@code !channel.isActive()}) handle that
+ * independently.
+ *
+ * @see ReactorNettyClient#configureChannelPipelineHandlers()
+ */
+final class Http2ParentChannelExceptionHandler extends ChannelInboundHandlerAdapter {
+
+    static final String HANDLER_NAME = "cosmosH2ParentExceptionHandler";
+
+    private static final Logger logger = LoggerFactory.getLogger(Http2ParentChannelExceptionHandler.class);
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+        int activeStreams = getActiveStreamCount(ctx);
+        boolean channelActive = ctx.channel().isActive();
+
+        if (activeStreams == 0 || !channelActive) {
+            // No active streams OR channel already inactive — exception is noise
+            // (e.g., TCP RST from LB idle timeout, post-close cleanup).
+            if (logger.isDebugEnabled()) {
+                logger.debug(
+                    "Exception on HTTP/2 parent connection [id:{}, activeStreams={}, channelActive={}]",
+                    ctx.channel().id().asShortText(), activeStreams, channelActive, cause);
+            }
+        } else {
+            // Active streams on a live channel — exception may affect in-flight requests.
+            logger.warn(
+                "Exception on HTTP/2 parent connection [id:{}, activeStreams={}, channelActive={}]",
+                ctx.channel().id().asShortText(), activeStreams, channelActive, cause);
+        }
+    }
+
+    private static int getActiveStreamCount(ChannelHandlerContext ctx) {
+        try {
+            Http2FrameCodec codec = ctx.pipeline().get(Http2FrameCodec.class);
+            if (codec != null) {
+                return codec.connection().numActiveStreams();
+            }
+        } catch (Exception ignored) {
+            // Codec not available or connection already torn down
+        }
+        return -1;
+    }
+}

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java

@@ -166,6 +166,29 @@ private void configureChannelPipelineHandlers() {
                             "customHeaderCleaner",
                             new Http2ResponseHeaderCleanerHandler());
                     }
+
+                    // Install exception handler on the HTTP/2 parent (TCP) channel.
+                    // In H2, doOnConnected fires for stream (child) channels — channel.parent()
+                    // is the TCP connection. The parent pipeline has no ChannelOperationsHandler
+                    // (unlike H1.1), so TCP-level exceptions (RST, broken pipe) propagate to
+                    // Netty's TailContext. This handler aligns the logging with connection state:
+                    // DEBUG when the parent channel is idle or inactive, WARN when the channel
+                    // is active and in-flight streams may be impacted.
+                    Channel parent = connection.channel().parent();
+                    if (parent != null
+                        && parent.pipeline().get(Http2ParentChannelExceptionHandler.HANDLER_NAME) == null) {
+
+                        try {
+                            parent.pipeline().addLast(
+                                Http2ParentChannelExceptionHandler.HANDLER_NAME,
+                                new Http2ParentChannelExceptionHandler());
+                        } catch (IllegalArgumentException ignored) {
+                            // TOCTOU race: between the get()==null check above and addLast(),
+                            // a concurrent stream's doOnConnected may have installed the handler.
+                            // Since we always pass a fresh instance (never null, never shared
+                            // across pipelines), duplicate handler name is the only possible cause.
+                        }
+                    }
                 }));
         }
     }