+ * Uses {@code iptables DROP} to blackhole PING ACKs, verifying the handler closes + * the broken connection after consecutive PING timeouts and a subsequent request + * uses a new connection. + *
+ * Run in Docker with {@code --cap-add=NET_ADMIN} (group: {@code manual-http-network-fault}). + *
+ * Two transports exercise the same handler -- pick one per pipeline run via system properties: + *
+ * Without this fix, the same exception would have been classified as a generic + * {@code GATEWAY_ENDPOINT_UNAVAILABLE (10001)}, refresh-location would have run, + * the regional endpoint would have been marked down, and the retry would have + * either failed or landed on a different region depending on preferred-regions. + *
+ * Requires Docker with {@code --cap-add=NET_ADMIN} or Linux with sudo.
+ */
+ @Test(groups = {"manual-http-network-fault"}, timeOut = TEST_TIMEOUT)
+ public void inFlightReadRetriesInSameRegionAfterPingClose() throws Exception {
+ // Short interval + timeout for fast detection
+ System.setProperty("COSMOS.HTTP2_PING_INTERVAL_IN_SECONDS", "1");
+ System.setProperty("COSMOS.HTTP2_PING_TIMEOUT_IN_SECONDS", "2");
+ System.setProperty("COSMOS.HTTP2_PING_HEALTH_ENABLED", "true");
+ // Override threshold to 2 for faster test (default=5 aligned with Rust SDK)
+ System.setProperty("COSMOS.HTTP2_PING_FAILURE_THRESHOLD", "2");
+
+ // Lifted out of the try so the finally-block cleanup can reach it whether or not
+ // the iptables ADD ran -- finally needs the exact -D form of whatever -A we installed.
+ String iptablesDelete = null;
+ Thread iptablesRemovalThread = null;
+
+ try {
+ safeClose(this.client);
+
+ // Single-connection pool via Http2ConnectionConfig API
+ GatewayConnectionConfig gwConfig = new GatewayConnectionConfig();
+ gwConfig.getHttp2ConnectionConfig()
+ .setEnabled(true)
+ .setMaxConnectionPoolSize(1)
+ .setMinConnectionPoolSize(1);
+
+ this.client = new CosmosClientBuilder()
+ .endpoint(TestConfigurations.HOST)
+ .key(TestConfigurations.MASTER_KEY)
+ .consistencyLevel(ConsistencyLevel.SESSION)
+ .gatewayMode(gwConfig)
+ .buildAsyncClient();
+ this.cosmosAsyncContainer = getSharedMultiPartitionCosmosContainerWithIdAsPartitionKey(this.client);
+
+ // Warm-up read -- establish the H2 connection and capture diagnostics so we can
+ // (a) prove H2 was actually negotiated, (b) discover the regional endpoint host
+ // for IP-scoped iptables targeting on the Compute / port-443 path, and
+ // (c) capture the initial channel-id and endpoint host to compare against
+ // the recovery response.
+ CosmosItemResponse
+ * Parses the diagnostics JSON rather than substring-scanning toString()
+ * so that a future change to JSON formatting can't silently break the
+ * test.
+ */
+ private String extractParentChannelId(CosmosDiagnostics diagnostics) throws JsonProcessingException {
+ ObjectNode node = (ObjectNode) Utils.getSimpleObjectMapper().readTree(diagnostics.toString());
+ JsonNode gwStats = node.get("gatewayStatisticsList");
+ if (gwStats != null && gwStats.isArray()) {
+ for (int i = gwStats.size() - 1; i >= 0; i--) {
+ JsonNode stat = gwStats.get(i);
+ if (stat.has("parentChannelId")) {
+ String id = stat.get("parentChannelId").asText();
+ if (id != null && !id.isEmpty() && !"null".equals(id)) {
+ return id;
+ }
+ }
+ }
+ }
+ throw new AssertionError("Could not extract parentChannelId from diagnostics: " + diagnostics);
+ }
+
+ /**
+ * Runs a shell command and throws on non-zero exit. The test must fail loudly
+ * if iptables setup fails (e.g., missing NET_ADMIN capability), rather than
+ * silently continuing without network fault injection and reporting a
+ * misleading assertion failure downstream. Cleanup-only callers in
+ * {@code finally} blocks can swallow the exception locally.
+ */
+ private static void execCommand(String command) throws Exception {
+ Process p = Runtime.getRuntime().exec(new String[]{"bash", "-c", command});
+ int exit = p.waitFor();
+ if (exit != 0) {
+ java.io.BufferedReader reader = new java.io.BufferedReader(
+ new java.io.InputStreamReader(p.getErrorStream()));
+ StringBuilder sb = new StringBuilder();
+ String line;
+ while ((line = reader.readLine()) != null) {
+ sb.append(line);
+ }
+ String stderr = sb.toString().trim();
+ logger.warn("Command '{}' exited with code {}: {}", command, exit, stderr);
+ throw new RuntimeException("Command failed (exit=" + exit + "): " + command + " -- " + stderr);
+ }
+ }
+}
diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyHttp2PingCloseTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyHttp2PingCloseTest.java
new file mode 100644
index 000000000000..acc4a2e834dc
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/implementation/ClientRetryPolicyHttp2PingCloseTest.java
@@ -0,0 +1,155 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.cosmos.implementation;
+
+import com.azure.cosmos.BridgeInternal;
+import com.azure.cosmos.CosmosException;
+import com.azure.cosmos.ThrottlingRetryOptions;
+import com.azure.cosmos.implementation.http.Http2PingTimeoutChannelClosedException;
+import com.azure.cosmos.implementation.perPartitionAutomaticFailover.GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover;
+import com.azure.cosmos.implementation.perPartitionCircuitBreaker.GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker;
+import com.azure.cosmos.implementation.routing.RegionalRoutingContext;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+import reactor.core.publisher.Mono;
+
+import java.net.URI;
+
+import static com.azure.cosmos.implementation.TestUtils.mockDiagnosticsClientContext;
+
+/**
+ * Verifies the {@link ClientRetryPolicy} H3 branch that handles channel
+ * closures driven by {@code Http2PingHandler} consecutive PING ACK timeouts
+ * (sub-status {@code GATEWAY_HTTP2_PING_TIMEOUT_CHANNEL_CLOSED} = 10006).
+ *
+ * Invariant under test: a PING-driven local transport failure MUST NOT cause
+ * {@code markEndpointUnavailableForRead} or {@code markEndpointUnavailableForWrite}
+ * to be invoked on {@link GlobalEndpointManager}. The remote gateway is not
+ * known to be unhealthy; we route via {@code shouldRetryOnGatewayTimeout}
+ * (bounded same-region retry for safely-retriable reads; noRetry for writes)
+ * instead of the GATEWAY_ENDPOINT_UNAVAILABLE path that does mark the
+ * endpoint down.
+ */
+public class ClientRetryPolicyHttp2PingCloseTest {
+
+ private static final int ITERATIONS = 10;
+
+ @Test(groups = "unit")
+ public void pingTimeoutClose_onRead_retriesInRegionWithoutEndpointMarkDown() throws Exception {
+ ThrottlingRetryOptions retryOptions = new ThrottlingRetryOptions();
+ GlobalEndpointManager endpointManager = Mockito.mock(GlobalEndpointManager.class);
+ GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker globalPartitionEndpointManagerForPerPartitionCircuitBreaker
+ = Mockito.mock(GlobalPartitionEndpointManagerForPerPartitionCircuitBreaker.class);
+ GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover globalPartitionEndpointManagerForPerPartitionAutomaticFailover
+ = Mockito.mock(GlobalPartitionEndpointManagerForPerPartitionAutomaticFailover.class);
+
+ Mockito.doReturn(new RegionalRoutingContext(new URI("http://localhost")))
+ .when(endpointManager).resolveServiceEndpoint(Mockito.any(RxDocumentServiceRequest.class));
+ Mockito.doReturn(Mono.empty()).when(endpointManager).refreshLocationAsync(Mockito.eq(null), Mockito.eq(false));
+ Mockito.doReturn(2).when(endpointManager).getPreferredLocationCount();
+
+ ClientRetryPolicy clientRetryPolicy = new ClientRetryPolicy(
+ mockDiagnosticsClientContext(),
+ endpointManager,
+ true,
+ retryOptions,
+ globalPartitionEndpointManagerForPerPartitionCircuitBreaker,
+ globalPartitionEndpointManagerForPerPartitionAutomaticFailover,
+ false);
+
+ CosmosException cosmosException = buildPingTimeoutCloseCosmosException();
+
+ RxDocumentServiceRequest dsr = RxDocumentServiceRequest.createFromName(mockDiagnosticsClientContext(),
+ OperationType.Read, "/dbs/db/colls/col/docs/docId", ResourceType.Document);
+ dsr.requestContext = new DocumentServiceRequestContext();
+ clientRetryPolicy.onBeforeSendRequest(dsr);
+
+ for (int i = 0; i < ITERATIONS; i++) {
+ Mono
+ * Covers state transitions that do not require advancing real time -- the handler's
+ * timeout / threshold logic reads {@code System.nanoTime()}, so the tests sidestep
+ * the clock by pre-setting {@code pingOutstandingSinceNanos} (or
+ * {@code lastActivityNanos}) via reflection:
+ *
+ * Installed at the head of each H2 child-stream pipeline by
+ * {@code ReactorNettyClient}'s {@code .doOnRequest(...)} hook. When the parent (TCP)
+ * channel is closed by {@link Http2PingHandler} after consecutive PING-ACK timeouts
+ * or consecutive PING-send failures, the H2 multiplex codec propagates
+ * {@code channelInactive} to every child stream.
+ * This handler fires first (head-of-pipeline), inspects the parent channel's
+ * {@link Http2PingHandler#PING_TIMEOUT_CLOSED} attribute, and on match fires
+ * {@code exceptionCaught} with the typed exception before the default close
+ * path reaches reactor-netty's {@code HttpClientOperations}. That makes the in-flight
+ * request's response {@code Mono} fail with the typed exception (which the rest of the
+ * stack maps to {@code GATEWAY_HTTP2_PING_TIMEOUT_CHANNEL_CLOSED}) instead of a bare
+ * {@code PrematureCloseException}, which would otherwise trigger region mark-down in
+ * {@code ClientRetryPolicy}.
+ *
+ * The handler is stateless and marked {@link ChannelHandler.Sharable}, so a single
+ * JVM-wide {@link #INSTANCE} is reused across all H2 child streams.
+ *
+ * For non-H2 channels (parent is {@code null}) this handler is never installed; the
+ * install site in {@code ReactorNettyClient} gates on {@code ch.parent() != null}.
+ */
+@ChannelHandler.Sharable
+final class Http2PingCloseRewrapHandler extends ChannelInboundHandlerAdapter {
+
+ static final String HANDLER_NAME = "cosmos.http2PingCloseRewrap";
+
+ static final Http2PingCloseRewrapHandler INSTANCE = new Http2PingCloseRewrapHandler();
+
+ private Http2PingCloseRewrapHandler() {}
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ Channel parent = ctx.channel().parent();
+ if (parent != null && Boolean.TRUE.equals(parent.attr(Http2PingHandler.PING_TIMEOUT_CLOSED).get())) {
+ // Fire BEFORE delegating to super.channelInactive so reactor-netty's
+ // HttpClientOperations.exceptionCaught fails the response Mono with the
+ // typed exception, beating its own onInboundClose(PrematureCloseException).
+ ctx.fireExceptionCaught(new Http2PingTimeoutChannelClosedException(
+ "HTTP/2 connection closed by PING keepalive after consecutive PING-ACK timeouts or PING-send failures.",
+ null));
+ }
+ super.channelInactive(ctx);
+ }
+}
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java
new file mode 100644
index 000000000000..82782d14c71b
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingHandler.java
@@ -0,0 +1,289 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.implementation.http;
+
+import com.azure.cosmos.Http2ConnectionConfig;
+import com.azure.cosmos.implementation.Configs;
+import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelDuplexHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http2.DefaultHttp2PingFrame;
+import io.netty.handler.codec.http2.Http2PingFrame;
+import io.netty.util.AttributeKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Manual HTTP/2 PING keepalive handler installed on the parent (TCP) channel.
+ *
+ * Sends PING frames when the connection is idle for longer than the configured interval,
+ * preventing L7 middleboxes (NAT gateways, firewalls, load balancers) from silently
+ * reaping the connection. Modeled after Rust SDK's hyper-based HTTP/2 PING approach.
+ *
+ * When a PING ACK is not received within the configured timeout, the handler closes
+ * the connection. The connection pool will then create a fresh replacement. This is
+ * aligned with the Rust SDK where hyper kills connections on PING timeout and the
+ * shard health sweep replaces them.
+ *
+ * Why manual instead of reactor-netty native {@code pingAckTimeout}? Native PING requires
+ * built-in {@code maxIdleTime} handling, which is bypassed when a custom
+ * {@code evictionPredicate} is configured (reactor-netty 1.2.13).
+ *
+ * Note: the Cosmos DB Proxy (Standard Gateway, ThinClient) uses nghttp2 which auto-ACKs
+ * PINGs per RFC 9113 §6.7. The SQLx Mux uses a custom frame parser that currently rejects
+ * PING with PROTOCOL_ERROR — but SQLx does not negotiate H2 via ALPN, so clients never
+ * speak H2 to it.
+ */
+public class Http2PingHandler extends ChannelDuplexHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(Http2PingHandler.class);
+
+ private static final String HANDLER_NAME = "cosmos.http2PingHandler";
+
+ /**
+ * Marker attribute set on the parent (TCP) channel immediately before {@link
+ * ChannelHandlerContext#close()} when this handler closes the channel due to
+ * consecutive PING ACK timeouts (or PING send failures). Downstream code can
+ * use this to discriminate a PING-driven close from other channel closures.
+ *
+ * Consumed by {@link Http2PingCloseRewrapHandler}, a {@code @Sharable} handler
+ * installed at the head of each H2 child-stream pipeline by {@code
+ * ReactorNettyClient.doOnRequest(...)}. When the parent channel closes with this
+ * attribute set, the rewrap handler fires {@code exceptionCaught} with a typed
+ * {@link Http2PingTimeoutChannelClosedException} so the in-flight request's
+ * response {@code Mono} fails with the typed exception (instead of a bare
+ * {@code PrematureCloseException}); upstream {@code ClientRetryPolicy} then
+ * suppresses region mark-down for the corresponding subStatusCode.
+ */
+ public static final AttributeKey
+ * All three gates must pass:
+ *
+ * This is a local transport failure -- typically caused by NAT / load
+ * balancer idle reaping of an otherwise-healthy connection. The remote
+ * service (e.g. Cosmos DB Standard Gateway or ThinClient proxy) is NOT
+ * known to be unhealthy. Code paths that classify network failures (e.g.
+ * {@code com.azure.cosmos.implementation.ClientRetryPolicy}) should use this
+ * marker to suppress region mark-down / cross-region failover and instead
+ * retry against the same regional endpoint with a fresh connection.
+ *
+ * Extends {@link ClosedChannelException} so existing classifiers (notably
+ * {@code WebExceptionUtility.isNetworkFailure}) continue to recognize it as a
+ * network failure; discrimination from a generic closed channel is via {@code
+ * instanceof} (see {@code WebExceptionUtility.isHttp2PingTimeoutClose}).
+ *
+ * Note: {@link ClosedChannelException} only exposes a no-arg constructor, so
+ * the message is carried in a private field and surfaced via {@link
+ * #getMessage()}.
+ */
+public final class Http2PingTimeoutChannelClosedException extends ClosedChannelException {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String message;
+
+ public Http2PingTimeoutChannelClosedException(String message, Throwable cause) {
+ super();
+ this.message = message;
+ if (cause != null) {
+ initCause(cause);
+ }
+ }
+
+ @Override
+ public String getMessage() {
+ return message;
+ }
+}
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
index 2e362eadc084..e95989c5aa3c 100644
--- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/ReactorNettyClient.java
@@ -11,6 +11,7 @@
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http2.Http2MultiplexHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.resolver.DefaultAddressResolverGroup;
import io.netty.util.ReferenceCountUtil;
@@ -43,7 +44,7 @@
* HttpClient that is implemented using reactor-netty.
*/
public class ReactorNettyClient implements HttpClient {
- private static ImplementationBridgeHelpers.Http2ConnectionConfigHelper.Http2ConnectionConfigAccessor httpCfgAccessor() {
+ private static ImplementationBridgeHelpers.Http2ConnectionConfigHelper.Http2ConnectionConfigAccessor http2CfgAccessor() {
return ImplementationBridgeHelpers.Http2ConnectionConfigHelper.getHttp2ConnectionConfigAccessor();
}
@@ -141,7 +142,38 @@ private void configureChannelPipelineHandlers() {
.validateHeaders(true));
Http2ConnectionConfig http2Cfg = httpClientConfig.getHttp2ConnectionConfig();
- if (httpCfgAccessor().isEffectivelyEnabled(http2Cfg)) {
+
+ boolean isH2Enabled = http2CfgAccessor().isEffectivelyEnabled(http2Cfg);
+
+ if (isH2Enabled) {
+ this.httpClient = this.httpClient.doOnConnected(connection -> {
+ // Manual HTTP/2 PING keepalive -- sends PING frames when the connection is idle
+ // to prevent L7 middleboxes (NAT, firewalls, LBs) from reaping the connection.
+ // For H2, doOnConnected fires on the parent TCP channel when the connection
+ // is first established (State.CONFIGURED). Child stream channels emit
+ // STREAM_CONFIGURED, which does not trigger this callback. The parent-
+ // resolution and installIfAbsent guard below are defensive -- they correctly
+ // handle both parent and child channels if the reactor-netty contract
+ // ever changes.
+ //
+ // Gating is consolidated in Http2PingHandler.isPingHealthEffectivelyEnabled so
+ // the transport install site and the user-agent feature flag cannot drift.
+ // The handler also re-checks the kill-switch per tick so toggling it at
+ // runtime immediately stops/resumes PINGing on already-installed connections.
+ if (Http2PingHandler.isPingHealthEffectivelyEnabled(http2Cfg)) {
+ // Resolve to the parent (TCP) channel -- defensive in case reactor-netty
+ // ever invokes this callback for a child stream channel.
+ Channel ch = connection.channel();
+ Channel parent = ch.parent() != null ? ch.parent() : ch;
+ if (parent.pipeline().get(Http2MultiplexHandler.class) != null) {
+ Http2PingHandler.installIfAbsent(parent,
+ Configs.getHttp2PingIntervalInSeconds(),
+ Configs.getHttp2PingTimeoutInSeconds(),
+ Configs.getHttp2PingFailureThreshold());
+ }
+ }
+ });
+
this.httpClient = this.httpClient
.secure(sslContextSpec ->
sslContextSpec.sslContext(
@@ -153,18 +185,25 @@ private void configureChannelPipelineHandlers() {
.http2Settings(settings -> settings
.initialWindowSize(1024 * 1024) // 1MB initial window size
.maxFrameSize(64 * 1024) // 64KB max frame size
- .maxConcurrentStreams(httpCfgAccessor().getEffectiveMaxConcurrentStreams(http2Cfg)) // Increased from default 30
+ .maxConcurrentStreams(http2CfgAccessor().getEffectiveMaxConcurrentStreams(http2Cfg)) // Increased from default 30
)
.doOnConnected((connection -> {
// The response header clean up pipeline is being added due to an error getting when calling gateway:
// java.lang.IllegalArgumentException: a header value contains prohibited character 0x20 at index 0 for 'x-ms-serviceversion', there is whitespace in the front of the value.
// validateHeaders(false) does not work for http2
ChannelPipeline channelPipeline = connection.channel().pipeline();
- if (channelPipeline.get("reactor.left.httpCodec") != null) {
- channelPipeline.addAfter(
- "reactor.left.httpCodec",
- "customHeaderCleaner",
- new Http2ResponseHeaderCleanerHandler());
+ if (channelPipeline.get("reactor.left.httpCodec") != null
+ && channelPipeline.get("customHeaderCleaner") == null) {
+ try {
+ channelPipeline.addAfter(
+ "reactor.left.httpCodec",
+ "customHeaderCleaner",
+ new Http2ResponseHeaderCleanerHandler());
+ } catch (IllegalArgumentException ignored) {
+ // TOCTOU race: between the get()==null check above and addAfter(),
+ // a concurrent doOnConnected may have installed the handler.
+ // Duplicate handler name is the only possible cause.
+ }
}
// Install exception handler at the tail of the HTTP/2 parent (TCP)
@@ -184,7 +223,35 @@ private void configureChannelPipelineHandlers() {
// Duplicate handler name is the only possible cause.
}
}
- }));
+ }))
+ .doOnRequest((req, conn) -> {
+ // Install a @Sharable head-of-pipeline rewrap handler on each H2
+ // child-stream pipeline. When Http2PingHandler closes the parent
+ // (TCP) channel after consecutive PING-ACK timeouts or PING-send
+ // failures, the H2 multiplex codec propagates channelInactive to
+ // every child stream; the rewrap handler intercepts that and fires
+ // exceptionCaught with a typed
+ // Http2PingTimeoutChannelClosedException so reactor-netty's
+ // HttpClientOperations fails the response Mono with the typed
+ // exception (instead of bare PrematureCloseException). The rest of the
+ // stack maps that to GATEWAY_HTTP2_PING_TIMEOUT_CHANNEL_CLOSED so
+ // ClientRetryPolicy can suppress region mark-down.
+ //
+ // Gate on ch.parent() != null so this only runs on H2 child streams
+ // (H1.1 connections have null parent and never need the rewrap).
+ Channel ch = conn.channel();
+ if (ch.parent() != null && ch.pipeline().get(Http2PingCloseRewrapHandler.HANDLER_NAME) == null) {
+ try {
+ ch.pipeline().addFirst(
+ Http2PingCloseRewrapHandler.HANDLER_NAME,
+ Http2PingCloseRewrapHandler.INSTANCE);
+ } catch (IllegalArgumentException ignored) {
+ // TOCTOU race: between the get()==null check above and addFirst(),
+ // a concurrent doOnRequest may have installed the handler.
+ // Duplicate handler name is the only possible cause.
+ }
+ }
+ });
}
}
@@ -253,7 +320,8 @@ public Mono
+ *
+ * The interval-driven scheduling itself (that {@code maybeSendPing} is fired by the
+ * event loop every {@code pingIntervalNanos}) is exercised by the integration test
+ * {@code Http2PingKeepaliveTest} under Docker with real network fault injection.
+ */
+public class Http2PingHandlerTest {
+
+ @Test(groups = "unit")
+ public void ackWithMatchingPayload_resetsState() throws Exception {
+ Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ // Simulate an outstanding PING #5 that recorded one failure.
+ setField(handler, "pingsSent", 5);
+ setField(handler, "pingOutstandingSinceNanos", System.nanoTime());
+ setField(handler, "consecutiveFailures", 1);
+
+ channel.writeInbound(new DefaultHttp2PingFrame(5L, true));
+
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat((int) getField(handler, "consecutiveFailures")).isZero();
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void ackWithMismatchedPayload_doesNotResetState() throws Exception {
+ Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ long outstandingAt = System.nanoTime();
+ setField(handler, "pingsSent", 5);
+ setField(handler, "pingOutstandingSinceNanos", outstandingAt);
+ setField(handler, "consecutiveFailures", 2);
+
+ // Late ACK for an earlier PING #3 -- must NOT clear the failure counter
+ // that is tracking the currently outstanding PING #5.
+ channel.writeInbound(new DefaultHttp2PingFrame(3L, true));
+
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isEqualTo(outstandingAt);
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(2);
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void ackAfterTimeoutCleared_doesNotResetState() throws Exception {
+ // Reproduces the race fixed by the `pingOutstandingSinceNanos != 0` guard in
+ // channelRead: PING #5 is sent (pingsSent=5), timeout fires and sets
+ // pingOutstandingSinceNanos=0 + consecutiveFailures=1, then a late ACK for #5
+ // arrives BEFORE the next PING is sent (so pingsSent is still 5). Payload alone
+ // matches; the outstanding-flag guard is what prevents masking the failure.
+ Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ setField(handler, "pingsSent", 5);
+ setField(handler, "pingOutstandingSinceNanos", 0L);
+ setField(handler, "consecutiveFailures", 1);
+
+ channel.writeInbound(new DefaultHttp2PingFrame(5L, true));
+
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(1);
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void installIfAbsent_isIdempotent() {
+ EmbeddedChannel channel = new EmbeddedChannel();
+
+ Http2PingHandler.installIfAbsent(channel, 1, 1, 3);
+ Http2PingHandler.installIfAbsent(channel, 1, 1, 3);
+
+ long count = channel.pipeline().toMap().values().stream()
+ .filter(h -> h instanceof Http2PingHandler)
+ .count();
+ assertThat(count).isEqualTo(1);
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void constructor_clampsNonPositiveValues() throws Exception {
+ Http2PingHandler handler = new Http2PingHandler(-5, 0, -1);
+
+ assertThat((long) getField(handler, "pingIntervalNanos")).isEqualTo(TimeUnit.SECONDS.toNanos(1));
+ assertThat((long) getField(handler, "pingTimeoutNanos")).isEqualTo(TimeUnit.SECONDS.toNanos(1));
+ assertThat((int) getField(handler, "failureThreshold")).isEqualTo(1);
+ }
+
+ @Test(groups = "unit")
+ public void killSwitchOff_clearsOutstandingPingState() throws Exception {
+ // M1: When Configs.isHttp2PingHealthEnabled() flips to false, the next maybeSendPing
+ // tick must clear any in-flight PING bookkeeping. Otherwise toggling the kill-switch
+ // back on after a long dormant window would charge a spurious timeout from a stale
+ // pingOutstandingSinceNanos that has been "outstanding" for hours.
+ final String prop = "COSMOS.HTTP2_PING_HEALTH_ENABLED";
+ final String prior = System.getProperty(prop);
+ try {
+ Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ // Simulate an in-flight PING with a partial failure already accumulated.
+ setField(handler, "pingOutstandingSinceNanos", System.nanoTime());
+ setField(handler, "consecutiveFailures", 2);
+
+ // Flip kill-switch OFF and trigger the periodic tick.
+ System.setProperty(prop, "false");
+ ChannelHandlerContext ctx = channel.pipeline().firstContext();
+ invokeMaybeSendPing(handler, ctx);
+
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat((int) getField(handler, "consecutiveFailures")).isZero();
+
+ // Timer must still be alive so re-enabling resumes PINGing on the same connection.
+ assertThat((Object) getField(handler, "pingTask")).isNotNull();
+
+ channel.finishAndReleaseAll();
+ } finally {
+ if (prior == null) {
+ System.clearProperty(prop);
+ } else {
+ System.setProperty(prop, prior);
+ }
+ }
+ }
+
+ @Test(groups = "unit")
+ public void channelInactive_cancelsPingTask() throws Exception {
+ // M2: channelInactive must cancel the scheduled PING task. Otherwise the timer
+ // would keep firing on a dead connection until GC reclaims the handler.
+ Http2PingHandler handler = new Http2PingHandler(1, 1, 3);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ ScheduledFuture scheduled = (ScheduledFuture) getField(handler, "pingTask");
+ assertThat((Object) scheduled).isNotNull();
+ assertThat(scheduled.isCancelled()).isFalse();
+
+ // Closing the EmbeddedChannel fires channelInactive on the handler.
+ channel.close().syncUninterruptibly();
+
+ assertThat(scheduled.isCancelled()).isTrue();
+ assertThat((Object) getField(handler, "pingTask")).isNull();
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void writeFailure_incrementsConsecutiveFailuresAndClosesAtThreshold() throws Exception {
+ // Reviewer-flagged invariant (PR #49095): a failed PING write must count as a
+ // failed health probe. Otherwise a channel stuck in a state where writes always
+ // fail (e.g. H2 codec rejecting frames, stalled flow-control, queued
+ // ClosedChannelException not yet propagated) but channel.isActive() stays true
+ // would loop on every tick without ever reaching the close threshold.
+ final int threshold = 2;
+ Http2PingHandler handler = new Http2PingHandler(1, 1, threshold);
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+ // Insert an outbound handler at HEAD that fails every write. Outbound writes
+ // from Http2PingHandler flow head-ward, so this intercepts them and fails the
+ // promise synchronously -- producing the same listener-on-failure path the
+ // handler would see if a real H2 codec rejected the PING frame.
+ channel.pipeline().addFirst("failOnWrite", new ChannelOutboundHandlerAdapter() {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+ ReferenceCountUtil.release(msg);
+ promise.setFailure(new IOException("simulated write failure"));
+ }
+ });
+
+ ChannelHandlerContext ctx = channel.pipeline().context(handler);
+ long longAgo = System.nanoTime() - TimeUnit.SECONDS.toNanos(60);
+
+ // First tick: force idle -> PING send attempt -> write fails.
+ // Expect consecutiveFailures=1, channel still open, task still scheduled.
+ setField(handler, "lastActivityNanos", longAgo);
+ invokeMaybeSendPing(handler, ctx);
+ channel.runPendingTasks(); // run the write + listener on the embedded event loop
+
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(1);
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat(channel.isOpen()).isTrue();
+ assertThat((Object) getField(handler, "pingTask")).isNotNull();
+
+ // Second tick: write fails again -> threshold reached -> task cancelled, channel closed.
+ setField(handler, "lastActivityNanos", longAgo);
+ invokeMaybeSendPing(handler, ctx);
+ channel.runPendingTasks();
+
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(threshold);
+ assertThat((Object) getField(handler, "pingTask")).isNull();
+ assertThat(channel.isOpen()).isFalse();
+
+ channel.finishAndReleaseAll();
+ }
+
+ @Test(groups = "unit")
+ public void ackTimeout_incrementsConsecutiveFailuresAndClosesAtThreshold() throws Exception {
+ // Reviewer-flagged invariant (PR #49095): the primary operational PING-failure
+ // path -- a PING was sent, no ACK arrived within pingTimeoutNanos, and after
+ // `failureThreshold` consecutive such timeouts the channel must be closed.
+ // Time is sidestepped by pre-setting pingOutstandingSinceNanos to a value
+ // older than pingTimeoutNanos so maybeSendPing enters the timeout branch on
+ // each invocation.
+ final int threshold = 2;
+ Http2PingHandler handler = new Http2PingHandler(1, 1, threshold); // interval=1s, timeout=1s
+ EmbeddedChannel channel = new EmbeddedChannel(handler);
+ ChannelHandlerContext ctx = channel.pipeline().context(handler);
+
+ long pastTimeout = System.nanoTime() - TimeUnit.SECONDS.toNanos(5);
+
+ // First tick: outstanding PING has aged past timeout -> failures=1, channel still open.
+ setField(handler, "pingsSent", 1);
+ setField(handler, "pingOutstandingSinceNanos", pastTimeout);
+ invokeMaybeSendPing(handler, ctx);
+ channel.runPendingTasks();
+
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(1);
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat(channel.isOpen()).isTrue();
+ assertThat((Object) getField(handler, "pingTask")).isNotNull();
+
+ // Second tick: simulate another outstanding PING that has aged past timeout
+ // -> failures=threshold -> task cancelled, channel closed.
+ setField(handler, "pingsSent", 2);
+ setField(handler, "pingOutstandingSinceNanos", pastTimeout);
+ invokeMaybeSendPing(handler, ctx);
+ channel.runPendingTasks();
+
+ assertThat((int) getField(handler, "consecutiveFailures")).isEqualTo(threshold);
+ assertThat((long) getField(handler, "pingOutstandingSinceNanos")).isZero();
+ assertThat((Object) getField(handler, "pingTask")).isNull();
+ assertThat(channel.isOpen()).isFalse();
+
+ channel.finishAndReleaseAll();
+ }
+
+ private static void invokeMaybeSendPing(Http2PingHandler handler, ChannelHandlerContext ctx) throws Exception {
+ Method m = Http2PingHandler.class.getDeclaredMethod("maybeSendPing", ChannelHandlerContext.class);
+ m.setAccessible(true);
+ m.invoke(handler, ctx);
+ }
+
+ private static Object getField(Object target, String name) throws Exception {
+ Field f = Http2PingHandler.class.getDeclaredField(name);
+ f.setAccessible(true);
+ return f.get(target);
+ }
+
+ private static void setField(Object target, String name, Object value) throws Exception {
+ Field f = Http2PingHandler.class.getDeclaredField(name);
+ f.setAccessible(true);
+ f.set(target, value);
+ }
+}
diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/GatewayReadConsistencyStrategySpyWireTest.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/GatewayReadConsistencyStrategySpyWireTest.java
index 5a2390408e20..99fc00b69f6f 100644
--- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/GatewayReadConsistencyStrategySpyWireTest.java
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/GatewayReadConsistencyStrategySpyWireTest.java
@@ -513,7 +513,7 @@ private HttpRequest executeQueryAndCapture(
CosmosQueryRequestOptions queryOptions) {
QueryFeedOperationState state = TestUtils.createDummyQueryFeedOperationState(
- ResourceType.Document, OperationType.Query, queryOptions, client);
+ ResourceType.Document, OperationType.Query, queryOptions, cosmosClient);
client.clearCapturedRequests();
client.queryDocuments(getCollectionLink(), "SELECT * FROM c", state, Document.class)
.blockFirst();
@@ -559,7 +559,7 @@ private HttpRequest executeReadManyAndCapture(
CosmosQueryRequestOptions queryOptions) {
QueryFeedOperationState state = TestUtils.createDummyQueryFeedOperationState(
- ResourceType.Document, OperationType.Query, queryOptions, client);
+ ResourceType.Document, OperationType.Query, queryOptions, cosmosClient);
client.clearCapturedRequests();
client.readManyByPartitionKeys(
Collections.singletonList(new PartitionKey(DOCUMENT_ID)),
diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java
index 0e84dde4c2cf..66b6e6314ad0 100644
--- a/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java
@@ -302,7 +302,7 @@ public CosmosAsyncDatabase getDatabase(String id) {
@BeforeSuite(groups = {"thinclient", "fast", "long", "direct", "multi-region", "multi-master", "flaky-multi-master", "emulator",
"emulator-vnext", "split", "query", "cfp-split", "circuit-breaker-misc-gateway", "circuit-breaker-misc-direct",
- "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong"}, timeOut = SUITE_SETUP_TIMEOUT)
+ "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault"}, timeOut = SUITE_SETUP_TIMEOUT)
public void beforeSuite() {
logger.info("beforeSuite Started");
@@ -353,7 +353,7 @@ private static DocumentCollection getInternalDocumentCollection(CosmosAsyncConta
@AfterSuite(groups = {"thinclient", "fast", "long", "direct", "multi-region", "multi-master", "flaky-multi-master",
"emulator", "split", "query", "cfp-split", "circuit-breaker-misc-gateway", "circuit-breaker-misc-direct",
- "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong"}, timeOut = SUITE_SHUTDOWN_TIMEOUT)
+ "circuit-breaker-read-all-read-many", "fi-multi-master", "long-emulator", "fi-thinclient-multi-region", "fi-thinclient-multi-master", "multi-region-strong", "manual-http-network-fault"}, timeOut = SUITE_SHUTDOWN_TIMEOUT)
public void afterSuite() {
logger.info("afterSuite Started");
diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/resources/log4j2-test.properties b/sdk/cosmos/azure-cosmos-tests/src/test/resources/log4j2-test.properties
index 116ca2900edd..b822e388c0f6 100644
--- a/sdk/cosmos/azure-cosmos-tests/src/test/resources/log4j2-test.properties
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/resources/log4j2-test.properties
@@ -7,3 +7,7 @@ appender.console.name = STDOUT
appender.console.type = Console
appender.console.layout.type = PatternLayout
appender.console.layout.pattern = %d %5X{pid} [%t] %-5p %c - %m%n
+
+# HTTP/2 PING handler -- enable DEBUG to see PING send/ACK/timeout/kill-switch activity
+logger.http2ping.name = com.azure.cosmos.implementation.http.Http2PingHandler
+logger.http2ping.level = debug
diff --git a/sdk/cosmos/azure-cosmos-tests/src/test/resources/manual-http-network-fault-testng.xml b/sdk/cosmos/azure-cosmos-tests/src/test/resources/manual-http-network-fault-testng.xml
new file mode 100644
index 000000000000..b6127237e422
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos-tests/src/test/resources/manual-http-network-fault-testng.xml
@@ -0,0 +1,16 @@
+
+
+ *
+ */
+ public static boolean isPingHealthEffectivelyEnabled(Http2ConnectionConfig http2Cfg) {
+ if (!Configs.isHttp2PingHealthEnabled()) {
+ return false;
+ }
+ if (Configs.getHttp2PingIntervalInSeconds() <= 0) {
+ return false;
+ }
+ return ImplementationBridgeHelpers.Http2ConnectionConfigHelper
+ .getHttp2ConnectionConfigAccessor()
+ .isEffectivelyEnabled(http2Cfg);
+ }
+
+ /**
+ * Installs the PING handler on the parent H2 channel if not already installed.
+ * For reactor-netty 1.2.x, {@code doOnConnected} fires on the parent TCP channel
+ * (State.CONFIGURED) and not on child stream channels (STREAM_CONFIGURED), so the
+ * input {@code channel} here is normally the parent already. The parent resolution
+ * and idempotency guard below are defensive in case that contract ever changes.
+ *
+ * @param channel the channel from doOnConnected (normally the parent, but child-safe)
+ * @param pingIntervalSeconds PING interval in seconds
+ * @param pingTimeoutSeconds PING ACK timeout in seconds
+ * @param failureThreshold consecutive timeouts before closing
+ */
+ public static void installIfAbsent(Channel channel, int pingIntervalSeconds, int pingTimeoutSeconds,
+ int failureThreshold) {
+ Channel parent = channel.parent() != null ? channel.parent() : channel;
+ if (parent.pipeline().get(HANDLER_NAME) == null) {
+ try {
+ parent.pipeline().addLast(HANDLER_NAME,
+ new Http2PingHandler(pingIntervalSeconds, pingTimeoutSeconds, failureThreshold));
+ } catch (IllegalArgumentException ignored) {
+ // Duplicate -- race between concurrent streams, benign
+ }
+ }
+ }
+}
diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingTimeoutChannelClosedException.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingTimeoutChannelClosedException.java
new file mode 100644
index 000000000000..fc28a056f3dc
--- /dev/null
+++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/http/Http2PingTimeoutChannelClosedException.java
@@ -0,0 +1,47 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+package com.azure.cosmos.implementation.http;
+
+import java.nio.channels.ClosedChannelException;
+
+/**
+ * Marker exception raised when {@link Http2PingHandler} closes an HTTP/2 parent
+ * channel after consecutive PING ACK timeouts (or PING send failures) crossed
+ * the configured threshold.
+ *