Split internalServerErrorCount by wire-visibility (#3297)

* Split internalServerErrorCount by wire-visibility

nettyMetrics.internalServerErrorCount previously incremented
unconditionally whenever a generic (non-RestServiceException,
non-client-termination) exception was handled in
NettyResponseChannel#getErrorResponse, even when the constructed 500
response could never actually be written to the client because
response metadata (e.g. a 200) had already been committed for a
streamed response.

This conflated two different failure modes under one counter/alert:
- a 500 that actually reached the client on the wire
- a post-commit failure where the client instead sees a force-closed
  connection after a partially delivered 200

Add internalServerErrorAfterResponseCommittedCount to track the
second case separately, so internalServerErrorCount now reflects only
errors that were actually written to the wire. The sum of the two
new counters equals what internalServerErrorCount counted before this
change.

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

* Revert metric split; track offline-service 500s after response commit instead

Replace the previous internalServerErrorCount /
internalServerErrorAfterResponseCommittedCount split with the original,
simpler behavior: internalServerErrorCount is incremented
unconditionally for generic internal errors again, exactly as before
this PR, so no existing alerting/dashboards keyed on it need to change.

A 500 that could not be delivered to the client because response
metadata (e.g. a 200) was already committed for a streamed GET is
still a case worth tracking - but only distinctly for known offline
(e.g. composite router secondary/parity-check) callers, since those
already-committed-response drops for that traffic are expected and
otherwise indistinguishable from a genuine client-facing incident.

Add a new netty.server.offline.service.ids config listing the
x-ambry-service-id values (as configured for those callers, e.g. in
the composite router config) that identify offline traffic. When a
request from one of those service IDs hits the already-committed-500
case, increment a new, additive NettyMetrics#offlineInternalServerErrorOnlyCount
counter (OfflineInternalServerErrorOnlyCount.Count.rrd) instead of
introducing a second alerting metric.

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

* Fix TOCTOU race in offline-metric detection

Read responseMetadataWriteInitiated after the CAS attempt in
maybeSendErrorResponse(), not before. A concurrent writer (e.g. a
router content-write callback) can commit response metadata on
another thread between the earlier snapshot and our own CAS attempt,
which could cause the offline metric to be silently skipped even
though a response was already committed to the client.

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

* Add offline-only 503 (ServiceUnavailable) tracking, mirroring the 500 case

Adds offlineServiceUnavailableOnlyCount, incremented under the same
conditions as offlineInternalServerErrorOnlyCount but for a genuine
ServiceUnavailable (503) that never reached the wire because response
metadata was already committed, for a request from a configured
offline service ID. Host-level-throttled drops are excluded, matching
how the existing serviceUnavailableErrorCount already excludes them
from polluting SLO dashboards.

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

* Add offlineHostLevelThrottledOnlyCount for offline host-level-throttled drops

Mirrors offlineInternalServerErrorOnlyCount/offlineServiceUnavailableOnlyCount:
tracks host-level-throttled 503 drops that never reached the wire because
response metadata was already committed, for a request from a configured
offline service ID. Kept as its own counter, matching how hostLevelThrottledCount
is already tracked separately from serviceUnavailableErrorCount.

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

* Trim and drop empty entries when parsing netty.server.offline.service.ids

Previously an unconfigured (or trailing-comma) value produced a set
containing a single empty string instead of a truly empty set, so the
isEmpty() fast-path in isOfflineServiceRequest() was never hit by
default. A stray space after a comma also produced an id that could
never match via exact Set.contains(...). Trim each entry and drop
empties so both cases are handled correctly.

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

---------

Co-authored-by: Sophie Guo <sopguo@sopguo-mn2.linkedin.biz>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: 3 people

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<String> 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, "");

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);
+  }
+}

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 =

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<String, String> 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);
       }