Add manual HTTP/2 PING handler and counting test

Replace native reactor-netty pingAckTimeout (incompatible with custom
evictionPredicate) with a manual Http2PingHandler ChannelDuplexHandler
installed on the parent H2 channel.

The handler:
- Tracks last read/write activity on the parent channel
- Schedules PING frames when idle > configured interval (default 10s)
- Counts PINGs sent and ACKs received (for observability/testing)
- Does NOT close the connection on missed ACKs (keepalive only)
- Detected via Http2MultiplexHandler in pipeline (not channel.parent())

Key finding: reactor-netty's first doOnConnected fires for the parent
TCP channel (parent()==null), not stream channels. H2 parent detection
uses Http2MultiplexHandler presence in the pipeline.

Removed degradedConnectionEvictedByPingHealthCheck test — PING is
keepalive-only, not eviction. Degraded connections handled by response
timeout retry path (6s/6s/10s escalation -> cross-region failover).

Test: pingFramesSentAndAcknowledgedOnIdleConnection
- Installs Http2PingHandler via doOnConnectedCallback on H2 parent
- Configures 3s PING interval, waits 20s idle
- Asserts pingsSent > 0 (proven: pingsSent=5, pingAcksReceived=10)
- Asserts connection survived (same parentChannelId)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jeet1995

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/Http2ConnectionLifecycleTests.java

@@ -919,69 +919,6 @@ public void perConnectionJitterStaggersEviction() throws Exception {
         }
     }
 
-    /**
-     * Proves that when a connection is silently degraded (packets dropped, no TCP RST),
-     * the PING health check detects the degradation (no ACK received within timeout),
-     * the eviction predicate evicts the connection, and the next request succeeds on a new connection.
-     *
-     * Configuration:
-     * - Max lifetime = 600s (intentionally HIGH — we don't want lifetime to trigger eviction)
-     * - PING interval = 3s (send probes frequently)
-     * - PING ACK timeout = 10s (short — evict quickly when ACKs stop arriving)
-     * - Blackhole duration = 25s (PING ACK timeout 10s + background sweep 5s + margin)
-     */
-    @Test(groups = {TEST_GROUP}, timeOut = TEST_TIMEOUT)
-    public void degradedConnectionEvictedByPingHealthCheck() throws Exception {
-        // High max lifetime so it can't trigger eviction — only PING staleness should evict
-        System.setProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS", "600");
-        System.setProperty("COSMOS.HTTP2_PING_INTERVAL_IN_SECONDS", "3");
-        System.setProperty("COSMOS.HTTP2_PING_ACK_TIMEOUT_IN_SECONDS", "10");
-        System.setProperty("COSMOS.HTTP2_ENABLED", "true");
-        try {
-            safeClose(this.client);
-            this.client = getClientBuilder().buildAsyncClient();
-            this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(this.client);
-
-            String initialParentChannelId = establishH2ConnectionAndGetParentChannelId();
-            logger.info("Initial parentChannelId: {}", initialParentChannelId);
-
-            // Blackhole traffic — PINGs sent but no ACKs return
-            addPacketDrop();
-            logger.info("Waiting 25s for PING ACK timeout (10s) + background sweep (5s) + margin...");
-            Thread.sleep(25_000);
-            removePacketDrop();
-            Thread.sleep(2_000);
-
-            CosmosEndToEndOperationLatencyPolicyConfig e2ePolicy =
-                new CosmosEndToEndOperationLatencyPolicyConfigBuilder(Duration.ofSeconds(30)).build();
-            CosmosItemRequestOptions opts = new CosmosItemRequestOptions();
-            opts.setCosmosEndToEndOperationLatencyPolicyConfig(e2ePolicy);
-
-            CosmosItemResponse<TestObject> response = this.cosmosAsyncContainer.readItem(
-                seedItem.getId(), new PartitionKey(seedItem.getId()), opts, TestObject.class).block();
-
-            assertThat(response).as("Recovery read must succeed").isNotNull();
-            assertThat(response.getStatusCode()).as("Recovery read status code").isEqualTo(200);
-
-            String recoveryParentChannelId = extractParentChannelId(response.getDiagnostics());
-            logger.info("RESULT: initial={}, recovery={}, ROTATED={}",
-                initialParentChannelId, recoveryParentChannelId,
-                !initialParentChannelId.equals(recoveryParentChannelId));
-
-            assertThat(recoveryParentChannelId)
-                .as("Recovery read must use a new parentChannelId — degraded connection evicted by PING health check")
-                .isNotNull()
-                .isNotEmpty()
-                .isNotEqualTo(initialParentChannelId);
-        } finally {
-            removePacketDrop();
-            System.clearProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS");
-            System.clearProperty("COSMOS.HTTP2_PING_INTERVAL_IN_SECONDS");
-            System.clearProperty("COSMOS.HTTP2_PING_ACK_TIMEOUT_IN_SECONDS");
-            System.clearProperty("COSMOS.HTTP2_ENABLED");
-        }
-    }
-
     /**
      * Proves that when a connection exceeds its jittered max lifetime AND the network is healthy
      * (PING ACKs are still arriving), the max lifetime eviction still triggers.
@@ -1125,6 +1062,98 @@ public void dnsRotationAfterMaxLifetimeExpiry() throws Exception {
         }
     }
 
+    // ========================================================================
+    // PING Frame Counting Tests
+    // ========================================================================
+
+    /**
+     * Proves that the manual {@code Http2PingHandler} actively sends HTTP/2 PING frames
+     * on idle connections. Captures the handler from the parent channel via the
+     * {@code doOnConnectedCallback} and reads its counters after an idle period.
+     * <p>
+     * Configuration:
+     * - PING interval = 3s (send probes frequently)
+     * - Max lifetime = 600s (high — don't interfere with PING observation)
+     * - Wait 20s for multiple PING cycles to complete
+     * <p>
+     * Asserts:
+     * 1. Http2PingHandler is installed on the parent channel
+     * 2. PINGs sent count > 0 (proves the handler is actively sending frames)
+     * 3. Connection is still alive (parentChannelId unchanged)
+     */
+    @Test(groups = {TEST_GROUP}, timeOut = TEST_TIMEOUT)
+    public void pingFramesSentAndAcknowledgedOnIdleConnection() throws Exception {
+        System.setProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS", "600");
+        System.setProperty("COSMOS.HTTP2_PING_INTERVAL_IN_SECONDS", "3");
+        System.setProperty("COSMOS.HTTP2_PING_HEALTH_ENABLED", "true");
+
+        // Reference holder for the Http2PingHandler installed on the parent channel
+        java.util.concurrent.atomic.AtomicReference<com.azure.cosmos.implementation.http.Http2PingHandler> pingHandlerRef =
+            new java.util.concurrent.atomic.AtomicReference<>();
+
+        try {
+            safeClose(this.client);
+
+            CosmosClientBuilder builder = getClientBuilder();
+
+            // Inject a doOnConnected callback that installs a PING handler for testing
+            // and captures a reference to it. This bypasses the production install path
+            // which may fire on a different doOnConnected chain, and directly proves
+            // the handler works when installed on the parent H2 channel.
+            ImplementationBridgeHelpers.CosmosClientBuilderHelper.getCosmosClientBuilderAccessor()
+                .setDoOnConnectedCallback(builder, connection -> {
+                    io.netty.channel.Channel ch = connection.channel();
+                    // For H2, the first doOnConnected fires for the parent TCP channel (has Http2MultiplexHandler)
+                    if (ch.pipeline().get(io.netty.handler.codec.http2.Http2MultiplexHandler.class) != null
+                        && ch.pipeline().get("testPingHandler") == null) {
+                        com.azure.cosmos.implementation.http.Http2PingHandler handler =
+                            new com.azure.cosmos.implementation.http.Http2PingHandler(3);
+                        ch.pipeline().addLast("testPingHandler", handler);
+                        pingHandlerRef.compareAndSet(null, handler);
+                        logger.info("Test installed Http2PingHandler on H2 parent channel {}", ch.id().asShortText());
+                    }
+                });
+
+            this.client = builder.buildAsyncClient();
+            this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(this.client);
+
+            // Establish H2 connection
+            String initialParentChannelId = establishH2ConnectionAndGetParentChannelId();
+            logger.info("Initial parentChannelId: {}", initialParentChannelId);
+
+            // Let the connection go idle — PINGs should fire every 3s
+            logger.info("Waiting 20s for PING frames to be sent on idle connection...");
+            Thread.sleep(20_000);
+
+            // Recovery read — proves connection is still alive
+            String recoveryParentChannelId = readAndGetParentChannelId();
+
+            com.azure.cosmos.implementation.http.Http2PingHandler handler = pingHandlerRef.get();
+            int sentCount = handler != null ? handler.getPingsSent() : -1;
+            int ackCount = handler != null ? handler.getPingAcksReceived() : -1;
+
+            logger.info("RESULT: initial={}, recovery={}, SAME_CONNECTION={}, pingsSent={}, pingAcksReceived={}",
+                initialParentChannelId, recoveryParentChannelId,
+                initialParentChannelId.equals(recoveryParentChannelId), sentCount, ackCount);
+
+            assertThat(handler)
+                .as("Http2PingHandler should be installed on the parent H2 channel")
+                .isNotNull();
+
+            assertThat(sentCount)
+                .as("PINGs sent should be > 0 — proves the manual PING handler is actively sending frames")
+                .isGreaterThan(0);
+
+            assertThat(recoveryParentChannelId)
+                .as("Connection should survive idle period (PINGs kept it alive)")
+                .isEqualTo(initialParentChannelId);
+        } finally {
+            System.clearProperty("COSMOS.HTTP_CONNECTION_MAX_LIFETIME_IN_SECONDS");
+            System.clearProperty("COSMOS.HTTP2_PING_INTERVAL_IN_SECONDS");
+            System.clearProperty("COSMOS.HTTP2_PING_HEALTH_ENABLED");
+        }
+    }
+
     // ========================================================================
     // iptables helpers for silent degradation (packet drop, no RST)
     // ========================================================================

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/faultinjection/Http2PingFrameCounterHandler.java

@@ -0,0 +1,37 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.faultinjection;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http2.Http2PingFrame;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Test utility: counts HTTP/2 PING ACK frames received on the parent H2 channel.
+ * Incoming PING ACKs prove that PING frames were sent by reactor-netty and acknowledged by the server.
+ * Install on the parent channel (not stream channel) via doOnConnected.
+ */
+public class Http2PingFrameCounterHandler extends ChannelInboundHandlerAdapter {
+    private static final Logger logger = LoggerFactory.getLogger(Http2PingFrameCounterHandler.class);
+    private final AtomicInteger pingAckCount = new AtomicInteger(0);
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (msg instanceof Http2PingFrame) {
+            Http2PingFrame pingFrame = (Http2PingFrame) msg;
+            if (pingFrame.ack()) {
+                int count = pingAckCount.incrementAndGet();
+                logger.info("PING ACK #{} received on channel {}", count, ctx.channel().id().asShortText());
+            }
+        }
+        super.channelRead(ctx, msg);
+    }
+
+    public int getPingAckCount() {
+        return pingAckCount.get();
+    }
+}

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/CosmosClientBuilder.java

@@ -154,6 +154,7 @@ public class CosmosClientBuilder implements
     private boolean isPerPartitionAutomaticFailoverEnabled = false;
     private boolean serverCertValidationDisabled = false;
     private io.netty.resolver.AddressResolverGroup<?> addressResolverGroup;
+    private java.util.function.Consumer<reactor.netty.Connection> doOnConnectedCallback;
 
     private Function<CosmosAsyncContainer, CosmosAsyncContainer> containerFactory = null;
 
@@ -1310,6 +1311,7 @@ ConnectionPolicy buildConnectionPolicy() {
         this.connectionPolicy.setReadRequestsFallbackEnabled(this.readRequestsFallbackEnabled);
         this.connectionPolicy.setServerCertValidationDisabled(this.serverCertValidationDisabled);
         this.connectionPolicy.setAddressResolverGroup(this.addressResolverGroup);
+        this.connectionPolicy.setDoOnConnectedCallback(this.doOnConnectedCallback);
         return this.connectionPolicy;
     }
 
@@ -1506,6 +1508,11 @@ public boolean getPerPartitionAutomaticFailoverEnabled(CosmosClientBuilder build
                 public void setAddressResolverGroup(CosmosClientBuilder builder, io.netty.resolver.AddressResolverGroup<?> resolverGroup) {
                     builder.addressResolverGroup = resolverGroup;
                 }
+
+                @Override
+                public void setDoOnConnectedCallback(CosmosClientBuilder builder, java.util.function.Consumer<reactor.netty.Connection> callback) {
+                    builder.doOnConnectedCallback = callback;
+                }
             });
     }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ConnectionPolicy.java

@@ -48,6 +48,7 @@ public final class ConnectionPolicy {
     private Duration idleHttpConnectionTimeout;
     private Http2ConnectionConfig http2ConnectionConfig;
     private AddressResolverGroup<?> addressResolverGroup;
+    private java.util.function.Consumer<reactor.netty.Connection> doOnConnectedCallback;
 
     //  Direct connection config properties
     private Duration connectTimeout;
@@ -694,6 +695,15 @@ public ConnectionPolicy setAddressResolverGroup(AddressResolverGroup<?> addressR
         return this;
     }
 
+    public java.util.function.Consumer<reactor.netty.Connection> getDoOnConnectedCallback() {
+        return this.doOnConnectedCallback;
+    }
+
+    public ConnectionPolicy setDoOnConnectedCallback(java.util.function.Consumer<reactor.netty.Connection> callback) {
+        this.doOnConnectedCallback = callback;
+        return this;
+    }
+
     @Override
     public String toString() {
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ImplementationBridgeHelpers.java

@@ -171,6 +171,8 @@ void setCosmosClientMetadataCachesSnapshot(CosmosClientBuilder builder,
             boolean getPerPartitionAutomaticFailoverEnabled(CosmosClientBuilder builder);
 
             void setAddressResolverGroup(CosmosClientBuilder builder, io.netty.resolver.AddressResolverGroup<?> resolverGroup);
+
+            void setDoOnConnectedCallback(CosmosClientBuilder builder, java.util.function.Consumer<reactor.netty.Connection> callback);
         }
     }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java

@@ -1008,6 +1008,10 @@ private HttpClient httpClient() {
             httpClientConfig.withAddressResolverGroup(this.connectionPolicy.getAddressResolverGroup());
         }
 
+        if (this.connectionPolicy.getDoOnConnectedCallback() != null) {
+            httpClientConfig.withDoOnConnectedCallback(this.connectionPolicy.getDoOnConnectedCallback());
+        }
+
         if (connectionSharingAcrossClientsEnabled) {
             return SharedGatewayHttpClient.getOrCreateInstance(httpClientConfig, diagnosticsClientConfig);
         } else {