diff --git a/ambry-api/src/main/java/com/github/ambry/config/NettyConfig.java b/ambry-api/src/main/java/com/github/ambry/config/NettyConfig.java index b51e3c4c00..1e2f4179c4 100644 --- a/ambry-api/src/main/java/com/github/ambry/config/NettyConfig.java +++ b/ambry-api/src/main/java/com/github/ambry/config/NettyConfig.java @@ -14,9 +14,11 @@ package com.github.ambry.config; import com.github.ambry.rest.RestRequestService; +import com.github.ambry.rest.RestUtils; import java.util.Arrays; import java.util.HashSet; import java.util.Set; +import java.util.stream.Collectors; /** @@ -41,6 +43,7 @@ public class NettyConfig { public static final String NETTY_METRICS_STOP_WAIT_TIMEOUT_SECONDS = "netty.metrics.stop.wait.timeout.seconds"; public static final String NETTY_SERVER_CLOSE_DELAY_TIMEOUT_MS = "netty.server.close.delay.timeout.ms"; public static final String NETTY_ENABLE_ONE_HUNDRED_CONTINUE = "netty.enable.one.hundred.continue"; + public static final String NETTY_SERVER_OFFLINE_SERVICE_IDS = "netty.server.offline.service.ids"; /** * Number of netty boss threads. @@ -174,6 +177,17 @@ public class NettyConfig { @Default("false") public final boolean nettyEnableOneHundredContinue; + /** + * A comma separated list of {@link com.github.ambry.rest.RestUtils.Headers#SERVICE_ID} values that identify known + * offline (e.g. composite router secondary/parity-check) callers, as configured for those callers elsewhere (e.g. + * the composite router config). Used to attribute 500s that never reach the wire (because response metadata was + * already committed) to a dedicated offline-only metric instead of dropping that visibility entirely. Ids are + * trimmed of surrounding whitespace and empty entries are dropped, so surrounding spaces after a comma are fine. + */ + @Config(NETTY_SERVER_OFFLINE_SERVICE_IDS) + @Default("") + public final Set nettyServerOfflineServiceIds; + public NettyConfig(VerifiableProperties verifiableProperties) { nettyServerBossThreadCount = verifiableProperties.getInt(NETTY_SERVER_BOSS_THREAD_COUNT, 1); nettyServerIdleTimeSeconds = verifiableProperties.getInt(NETTY_SERVER_IDLE_TIME_SECONDS, 60); @@ -190,6 +204,14 @@ public NettyConfig(VerifiableProperties verifiableProperties) { Integer.MAX_VALUE); nettyServerDenyListedQueryParams = new HashSet<>( Arrays.asList(verifiableProperties.getString(NETTY_SERVER_DENY_LISTED_QUERY_PARAMS, "").split(","))); + // trim whitespace around each id and drop empty entries so an unconfigured (or trailing-comma) value + // yields a truly empty set rather than a set containing "" or " "-padded ids that can never match a + // real service id via isOfflineServiceRequest()'s exact Set.contains(...) check. + nettyServerOfflineServiceIds = Arrays.stream(verifiableProperties.getString(NETTY_SERVER_OFFLINE_SERVICE_IDS, "") + .split(",")) + .map(String::trim) + .filter(id -> !id.isEmpty()) + .collect(Collectors.toSet()); nettyMultipartPostMaxSizeBytes = verifiableProperties.getLongInRange(NETTY_MULTIPART_POST_MAX_SIZE_BYTES, 20 * 1024 * 1024, 0, Long.MAX_VALUE); nettyServerSslFactory = verifiableProperties.getString(SSL_FACTORY_KEY, ""); diff --git a/ambry-api/src/test/java/com/github/ambry/config/NettyConfigTest.java b/ambry-api/src/test/java/com/github/ambry/config/NettyConfigTest.java new file mode 100644 index 0000000000..38399aefff --- /dev/null +++ b/ambry-api/src/test/java/com/github/ambry/config/NettyConfigTest.java @@ -0,0 +1,44 @@ +/* + * Copyright 2026 LinkedIn Corp. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + */ +package com.github.ambry.config; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Properties; +import org.junit.Assert; +import org.junit.Test; + + +/** + * Tests for {@link NettyConfig}. + */ +public class NettyConfigTest { + + @Test + public void testOfflineServiceIdsDefaultIsEmpty() { + NettyConfig config = new NettyConfig(new VerifiableProperties(new Properties())); + Assert.assertEquals("Unconfigured netty.server.offline.service.ids should yield an empty set, not {\"\"}", + Collections.emptySet(), config.nettyServerOfflineServiceIds); + } + + @Test + public void testOfflineServiceIdsTrimsWhitespaceAndDropsEmptyEntries() { + Properties properties = new Properties(); + properties.setProperty(NettyConfig.NETTY_SERVER_OFFLINE_SERVICE_IDS, " service-a, service-b ,,service-c"); + NettyConfig config = new NettyConfig(new VerifiableProperties(properties)); + Assert.assertEquals("Ids should be trimmed and empty entries dropped", + new HashSet<>(Arrays.asList("service-a", "service-b", "service-c")), config.nettyServerOfflineServiceIds); + } +} diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMetrics.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMetrics.java index 9982b4c93d..4c49fcbf45 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyMetrics.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyMetrics.java @@ -131,8 +131,20 @@ public class NettyMetrics { public final Counter unauthorizedCount; public final Counter goneCount; public final Counter internalServerErrorCount; + // incremented when a generic internal-error 500 could not be delivered to the client because response metadata + // (e.g. a 200) had already been committed for a streamed GET, and the request's service ID identifies it as a + // known offline (e.g. composite router secondary/parity-check) caller. internalServerErrorCount is still + // incremented unconditionally for these, same as before; this is purely additive visibility. + public final Counter offlineInternalServerErrorOnlyCount; + // same as offlineInternalServerErrorOnlyCount above, but for a 503 that could not be delivered to the client. + // excludes host-level-throttled drops, matching how serviceUnavailableErrorCount itself excludes them. + public final Counter offlineServiceUnavailableOnlyCount; public final Counter serviceUnavailableErrorCount; public final Counter hostLevelThrottledCount; + // same as offlineServiceUnavailableOnlyCount above, but specifically for host-level-throttled drops that could + // not be delivered to the client, matching how hostLevelThrottledCount itself is tracked separately from + // serviceUnavailableErrorCount. + public final Counter offlineHostLevelThrottledOnlyCount; public final Counter insufficientCapacityErrorCount; public final Counter preconditionFailedErrorCount; public final Counter methodNotAllowedErrorCount; @@ -325,10 +337,16 @@ public NettyMetrics(MetricRegistry metricRegistry) { goneCount = metricRegistry.counter(MetricRegistry.name(NettyResponseChannel.class, "GoneCount")); internalServerErrorCount = metricRegistry.counter(MetricRegistry.name(NettyResponseChannel.class, "InternalServerErrorCount")); + offlineInternalServerErrorOnlyCount = metricRegistry.counter( + MetricRegistry.name(NettyResponseChannel.class, "OfflineInternalServerErrorOnlyCount")); + offlineServiceUnavailableOnlyCount = metricRegistry.counter( + MetricRegistry.name(NettyResponseChannel.class, "OfflineServiceUnavailableOnlyCount")); serviceUnavailableErrorCount = metricRegistry.counter(MetricRegistry.name(NettyResponseChannel.class, "ServiceUnavailableErrorCount")); hostLevelThrottledCount = metricRegistry.counter(MetricRegistry.name(NettyResponseChannel.class, "HostLevelThrottledCount")); + offlineHostLevelThrottledOnlyCount = metricRegistry.counter( + MetricRegistry.name(NettyResponseChannel.class, "OfflineHostLevelThrottledOnlyCount")); insufficientCapacityErrorCount = metricRegistry.counter(MetricRegistry.name(NettyResponseChannel.class, "InsufficientCapacityErrorCount")); preconditionFailedErrorCount = diff --git a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java index ea4c4fc99c..102f0e4663 100644 --- a/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java +++ b/ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java @@ -116,6 +116,10 @@ class NettyResponseChannel implements RestResponseChannel { // temp variable to hold the error response status which will be overwritten on responseStatus if the error response // was successfully sent private ResponseStatus errorResponseStatus = null; + // set alongside errorResponseStatus in getErrorResponse() when the 503 being built is a host-level-throttled + // drop rather than a genuine service-unavailable error, so callers can exclude it the same way + // serviceUnavailableErrorCount already does. + private boolean errorResponseIsHostLevelThrottled = false; /** * A {@link ChannelFutureListener} that closes the {@link Channel} which is @@ -593,10 +597,48 @@ private boolean maybeSendErrorResponse(Exception exception) { nettyMetrics.errorResponseProcessingTimeInMs.update(processingTime); } else { logger.debug("Could not send error response on channel {}", ctx.channel()); + // Check the flag *after* the write attempt above, not before: response metadata can be committed + // concurrently by a writer on another thread (e.g. a router content-write callback), so reading the flag + // before the CAS attempt above is racy and can miss an already-committed response that was in fact + // committed by that other writer just before/around our own CAS attempt failed. Reading it here is safe + // because by the time maybeWriteResponseMetadata() returns, the flag is guaranteed to be true if *any* + // writer (this one or a concurrent one) has ever successfully committed metadata. + if (responseMetadataWriteInitiated.get() && isOfflineServiceRequest()) { + if (errorResponseStatus == ResponseStatus.InternalServerError) { + // response metadata (e.g. a 200) was already committed to the client before this failure occurred, so the + // 500 constructed above never reached the wire. internalServerErrorCount was still incremented above, + // unconditionally, exactly as before this change. This is purely additive visibility into how often known + // offline (e.g. composite router secondary/parity-check) callers hit this already-committed-response case. + nettyMetrics.offlineInternalServerErrorOnlyCount.inc(); + } else if (errorResponseStatus == ResponseStatus.ServiceUnavailable) { + if (errorResponseIsHostLevelThrottled) { + // same as above, but the drop was host-level-throttled rather than a genuine service-unavailable + // failure; tracked separately, matching how hostLevelThrottledCount is kept separate from + // serviceUnavailableErrorCount. + nettyMetrics.offlineHostLevelThrottledOnlyCount.inc(); + } else { + // same as above, but for a genuine (non-throttler-driven) 503 that never reached the wire. + nettyMetrics.offlineServiceUnavailableOnlyCount.inc(); + } + } + } } return responseSent; } + /** + * @return {@code true} if the current request's {@link RestUtils.Headers#SERVICE_ID} matches one of the + * configured offline (e.g. composite router secondary/parity-check) service IDs. {@code false} if there is no + * current request, no service ID on it, or no offline service IDs are configured. + */ + private boolean isOfflineServiceRequest() { + if (request == null || nettyConfig.nettyServerOfflineServiceIds.isEmpty()) { + return false; + } + Object serviceId = request.getArgs().get(RestUtils.Headers.SERVICE_ID); + return serviceId != null && nettyConfig.nettyServerOfflineServiceIds.contains(serviceId.toString()); + } + /** * Provided a cause, returns an error response with the right status and error message. * @param cause the cause of the error. @@ -607,6 +649,7 @@ private FullHttpResponse getErrorResponse(Throwable cause) { RestServiceErrorCode restServiceErrorCode = null; String errReason = null; Map errHeaders = null; + errorResponseIsHostLevelThrottled = false; if (cause instanceof RestServiceException) { RestServiceException restServiceException = (RestServiceException) cause; restServiceErrorCode = restServiceException.getErrorCode(); @@ -618,6 +661,7 @@ private FullHttpResponse getErrorResponse(Throwable cause) { // here skips the ServiceUnavailable counter increment getHttpResponseStatus would do. nettyMetrics.hostLevelThrottledCount.inc(); status = HttpResponseStatus.SERVICE_UNAVAILABLE; + errorResponseIsHostLevelThrottled = true; } else { status = getHttpResponseStatus(errorResponseStatus); } diff --git a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java index ca65ba8fb9..d1fc9c7914 100644 --- a/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java +++ b/ambry-rest/src/test/java/com/github/ambry/rest/NettyResponseChannelTest.java @@ -701,6 +701,182 @@ public void clientEarlyTerminationTest() throws Exception { MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(cetMetricName).getCount()); } + /** + * Tests that {@link NettyMetrics#internalServerErrorCount} is incremented unconditionally for generic internal + * errors (same as before), and that {@link NettyMetrics#offlineInternalServerErrorOnlyCount} is additionally + * incremented only when a generic internal failure occurs after response metadata (e.g. a 200) has already been + * committed to the client AND the request's {@link RestUtils.Headers#SERVICE_ID} identifies it as a configured + * offline (e.g. composite router secondary/parity-check) caller. + */ + @Test + public void offlineInternalServerErrorOnlyCountTest() throws Exception { + String offlineServiceId = "ambry-offline-parity-checker"; + NettyConfig nettyConfigWithOfflineServiceId = new NettyConfig(new VerifiableProperties(new Properties() { + { + setProperty(NettyConfig.NETTY_SERVER_OFFLINE_SERVICE_IDS, offlineServiceId); + } + })); + String iseMetricName = MetricRegistry.name(NettyResponseChannel.class, "InternalServerErrorCount"); + String offlineIseMetricName = + MetricRegistry.name(NettyResponseChannel.class, "OfflineInternalServerErrorOnlyCount"); + long iseBeforeCount = MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount(); + long offlineIseBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount(); + + // generic exception before any response metadata has been sent -> internalServerErrorCount tracked, no offline + // metric change even though the request is from the configured offline service ID, since the 500 reached the + // wire. + HttpHeaders offlineServiceHeaders = new DefaultHttpHeaders().set(RestUtils.Headers.SERVICE_ID, offlineServiceId); + EmbeddedChannel channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.OnResponseCompleteWithNonRestException.toString(), + offlineServiceHeaders)); + HttpResponse response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.INTERNAL_SERVER_ERROR, response.status()); + if (!(response instanceof FullHttpResponse)) { + while (channel.readOutbound() != null) { + } + } + assertEquals("internalServerErrorCount should have been tracked", iseBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount()); + assertEquals("offlineInternalServerErrorOnlyCount should not have changed", offlineIseBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount()); + + // generic exception after response metadata (200) already committed, request from the configured offline + // service ID -> internalServerErrorCount is still tracked exactly as before, and offlineInternalServerErrorOnlyCount + // is additionally tracked since the 500 never reached the wire. + iseBeforeCount = MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount(); + offlineIseBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount(); + channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidway.toString(), + offlineServiceHeaders)); + response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("internalServerErrorCount should still have been tracked", iseBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount()); + assertEquals("offlineInternalServerErrorOnlyCount should have been tracked", offlineIseBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount()); + + // same already-committed-response scenario, but request is NOT from a configured offline service ID -> + // internalServerErrorCount is still tracked, but offlineInternalServerErrorOnlyCount should not change. + iseBeforeCount = MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount(); + offlineIseBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount(); + channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidway.toString(), null)); + response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("internalServerErrorCount should still have been tracked", iseBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(iseMetricName).getCount()); + assertEquals("offlineInternalServerErrorOnlyCount should not have changed", offlineIseBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineIseMetricName).getCount()); + } + + /** + * Tests that {@link NettyMetrics#offlineServiceUnavailableOnlyCount} is incremented only when a genuine (non + * host-level-throttled) 503 could not be delivered to the client because response metadata (e.g. a 200) had + * already been committed, AND the request's {@link RestUtils.Headers#SERVICE_ID} identifies it as a configured + * offline caller. Host-level-throttled drops, which share the 503 wire status, must be excluded, matching how + * {@link NettyMetrics#serviceUnavailableErrorCount} itself already excludes them; those are instead tracked via + * {@link NettyMetrics#offlineHostLevelThrottledOnlyCount}. + */ + @Test + public void offlineServiceUnavailableOnlyCountTest() throws Exception { + String offlineServiceId = "ambry-offline-parity-checker"; + NettyConfig nettyConfigWithOfflineServiceId = new NettyConfig(new VerifiableProperties(new Properties() { + { + setProperty(NettyConfig.NETTY_SERVER_OFFLINE_SERVICE_IDS, offlineServiceId); + } + })); + String offlineSuMetricName = + MetricRegistry.name(NettyResponseChannel.class, "OfflineServiceUnavailableOnlyCount"); + String offlineHltMetricName = + MetricRegistry.name(NettyResponseChannel.class, "OfflineHostLevelThrottledOnlyCount"); + HttpHeaders offlineServiceHeaders = new DefaultHttpHeaders().set(RestUtils.Headers.SERVICE_ID, offlineServiceId); + + // genuine ServiceUnavailable after response metadata (200) already committed, request from the configured + // offline service ID -> offlineServiceUnavailableOnlyCount is tracked since the 503 never reached the wire. + long offlineSuBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount(); + long offlineHltBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount(); + EmbeddedChannel channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidwayServiceUnavailable.toString(), + offlineServiceHeaders)); + HttpResponse response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("offlineServiceUnavailableOnlyCount should have been tracked", offlineSuBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount()); + assertEquals("offlineHostLevelThrottledOnlyCount should not have changed", offlineHltBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount()); + + // host-level-throttled drop after response metadata already committed, request from the configured offline + // service ID -> offlineServiceUnavailableOnlyCount should NOT change (throttler-driven drops are excluded from + // it), but offlineHostLevelThrottledOnlyCount should be tracked instead. + offlineSuBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount(); + offlineHltBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount(); + channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidwayHostLevelThrottled.toString(), + offlineServiceHeaders)); + response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("offlineServiceUnavailableOnlyCount should not have changed", offlineSuBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount()); + assertEquals("offlineHostLevelThrottledOnlyCount should have been tracked", offlineHltBeforeCount + 1, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount()); + + // same genuine-ServiceUnavailable-after-commit scenario, but request is NOT from a configured offline service + // ID -> offlineServiceUnavailableOnlyCount should not change. + offlineSuBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount(); + channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidwayServiceUnavailable.toString(), + null)); + response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("offlineServiceUnavailableOnlyCount should not have changed", offlineSuBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineSuMetricName).getCount()); + + // same host-level-throttled-after-commit scenario, but request is NOT from a configured offline service ID -> + // offlineHostLevelThrottledOnlyCount should not change. + offlineHltBeforeCount = + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount(); + channel = createEmbeddedChannel(nettyConfigWithOfflineServiceId); + channel.writeInbound( + RestTestUtils.createRequest(HttpMethod.GET, TestingUri.ResponseFailureMidwayHostLevelThrottled.toString(), + null)); + response = channel.readOutbound(); + assertEquals("Unexpected response status", HttpResponseStatus.OK, response.status()); + while (channel.readOutbound() != null) { + } + assertFalse("Channel is not closed at the remote end", channel.isActive()); + assertEquals("offlineHostLevelThrottledOnlyCount should not have changed", offlineHltBeforeCount, + MockNettyMessageProcessor.METRIC_REGISTRY.getCounters().get(offlineHltMetricName).getCount()); + } + /** * Tests that a recognizable client-termination exception without a network-boundary observation does not record a * per-container client abort. @@ -1365,6 +1541,17 @@ private EmbeddedChannel createEmbeddedChannel() { return new EmbeddedChannel(chunkedWriteHandler, processor); } + /** + * Same as {@link #createEmbeddedChannel()}, but with the given {@code nettyConfig} injected into the + * {@link MockNettyMessageProcessor}, instead of the default one. + */ + private EmbeddedChannel createEmbeddedChannel(NettyConfig nettyConfig) { + ChunkedWriteHandler chunkedWriteHandler = new ChunkedWriteHandler(); + MockNettyMessageProcessor processor = new MockNettyMessageProcessor(); + processor.setNettyConfig(nettyConfig); + return new EmbeddedChannel(chunkedWriteHandler, processor); + } + /** * Verifies any callbacks queued in the {@code processor}. * @param processor the {@link MockNettyMessageProcessor} that contains the callbacks that need to be verified. @@ -1885,6 +2072,17 @@ enum TestingUri { * Response sending fails midway through a write. */ ResponseFailureMidway, + /** + * Response sending fails midway through a write with a {@link RestServiceException} mapping to + * {@link ResponseStatus#ServiceUnavailable}. + */ + ResponseFailureMidwayServiceUnavailable, + /** + * Response sending fails midway through a write with a {@link RestServiceException} using + * {@link RestServiceErrorCode#HostLevelThrottled}, which shares the 503 wire status with + * {@link ResponseStatus#ServiceUnavailable} but should be excluded from the offline-only 503 metric. + */ + ResponseFailureMidwayHostLevelThrottled, /** * When this request is received, a response with {@link RestUtils.Headers#CONTENT_LENGTH} set is returned. * The value of the header {@link MockNettyMessageProcessor#CHUNK_COUNT_HEADER_NAME} is used to determine the number @@ -2127,6 +2325,24 @@ private void handleRequest(HttpRequest httpRequest) throws Exception { restResponseChannel.onResponseComplete(new Exception()); // this should close the channel and the test will check for that. break; + case ResponseFailureMidwayServiceUnavailable: + callback = new ChannelWriteCallback(); + callback.setFuture(restResponseChannel.write( + ByteBuffer.wrap(TestingUri.ResponseFailureMidwayServiceUnavailable.toString().getBytes()), callback)); + writeCallbacksToVerify.add(callback); + restResponseChannel.onResponseComplete( + new RestServiceException("service unavailable midway", RestServiceErrorCode.ServiceUnavailable)); + // this should close the channel and the test will check for that. + break; + case ResponseFailureMidwayHostLevelThrottled: + callback = new ChannelWriteCallback(); + callback.setFuture(restResponseChannel.write( + ByteBuffer.wrap(TestingUri.ResponseFailureMidwayHostLevelThrottled.toString().getBytes()), callback)); + writeCallbacksToVerify.add(callback); + restResponseChannel.onResponseComplete( + new RestServiceException("host level throttled midway", RestServiceErrorCode.HostLevelThrottled)); + // this should close the channel and the test will check for that. + break; case ResponseWithContentLength: chunkCount = httpRequest.headers().getInt(CHUNK_COUNT_HEADER_NAME, -1); if (chunkCount == -1) {