[server][fc] Fix gRPC read stats recording successful reads as failures

GrpcOutboundStatsHandler chose between successRequest and errorRequest with:

    !ctx.hasError() && !responseStatus.equals(OK) ||
responseStatus.equals(NOT_FOUND)

Java evaluates && before ||, so this parses as
(!hasError && !status.equals(OK)) || status.equals(NOT_FOUND). A successful
value-found read (no error, status OK) evaluates to false and falls into the
else branch, recording the read via errorRequest(). That emits
Read.CallCount{Http.Response.StatusCode=200,
Venice.Response.StatusCodeCategory=fail}
and increments the error_request sensor for every successful gRPC read. The
same
broken condition also recorded genuine 400/500 errors and 429 throttles as
successes.

Parenthesize the condition to mirror the Netty StatsHandler: record a success
only when no handler flagged an error and the status is OK or NOT_FOUND;
record
TOO_MANY_REQUESTS as neither (a throttled request is not a server-side
failure);
record everything else as an error.

Added GrpcOutboundStatsHandlerTest covering value-found -> success (regression
guard), key-absent -> success, error-flagged-with-OK -> error, 400/500 ->
error,
and 429 -> neither.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: m-nagarajan

services/venice-server/src/main/java/com/linkedin/venice/listener/grpc/handlers/GrpcOutboundStatsHandler.java

@@ -38,10 +38,15 @@ public void processRequest(GrpcRequestContext ctx) {
 
     double elapsedTime = LatencyUtils.getElapsedTimeFromNSToMS(statsContext.getRequestStartTimeInNS());
 
-    if (!ctx.hasError() && !responseStatus.equals(HttpResponseStatus.OK)
-        || responseStatus.equals(HttpResponseStatus.NOT_FOUND)) {
+    /*
+     * Record a success only when no handler flagged an error and the response is OK (value found) or NOT_FOUND
+     * (key absent); otherwise record an error. TOO_MANY_REQUESTS is recorded as neither, since a throttled
+     * request is not a server-side failure. Mirrors the Netty StatsHandler categorization.
+     */
+    if (!ctx.hasError()
+        && (responseStatus.equals(HttpResponseStatus.OK) || responseStatus.equals(HttpResponseStatus.NOT_FOUND))) {
       statsContext.successRequest(serverHttpRequestStats, elapsedTime);
-    } else {
+    } else if (!responseStatus.equals(HttpResponseStatus.TOO_MANY_REQUESTS)) {
       statsContext.errorRequest(serverHttpRequestStats, elapsedTime);
     }
   }

services/venice-server/src/test/java/com/linkedin/venice/listener/grpc/handlers/GrpcOutboundStatsHandlerTest.java

@@ -0,0 +1,77 @@
+package com.linkedin.venice.listener.grpc.handlers;
+
+import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
+import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR;
+import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpResponseStatus.TOO_MANY_REQUESTS;
+import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.linkedin.venice.listener.ServerStatsContext;
+import com.linkedin.venice.listener.grpc.GrpcRequestContext;
+import com.linkedin.venice.protocols.VeniceServerResponse;
+import com.linkedin.venice.stats.AggServerHttpRequestStats;
+import com.linkedin.venice.stats.ServerHttpRequestStats;
+import io.grpc.stub.StreamObserver;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+
+public class GrpcOutboundStatsHandlerTest {
+  private static final String STORE_NAME = "test_store";
+
+  /*
+   * Each row is: the response status, whether a prior handler flagged an error, and the expected outcome -
+   * TRUE = recorded as success, FALSE = recorded as error, null = recorded as neither (throttled).
+   */
+  @DataProvider(name = "responseStatusCases", parallel = true)
+  public Object[][] responseStatusCases() {
+    return new Object[][] { { OK, false, Boolean.TRUE }, // value found -> success (regression guard for 200-as-fail)
+        { NOT_FOUND, false, Boolean.TRUE }, // key absent -> success
+        { OK, true, Boolean.FALSE }, // error flagged despite OK status -> error
+        { BAD_REQUEST, false, Boolean.FALSE }, // malformed request -> error
+        { INTERNAL_SERVER_ERROR, false, Boolean.FALSE }, // server failure -> error
+        { TOO_MANY_REQUESTS, false, null } }; // throttled -> neither success nor error
+  }
+
+  @Test(dataProvider = "responseStatusCases")
+  public void testRequestCategorization(HttpResponseStatus responseStatus, boolean hasError, Boolean expectSuccess) {
+    ServerHttpRequestStats storeStats = mock(ServerHttpRequestStats.class);
+    AggServerHttpRequestStats aggStats = mock(AggServerHttpRequestStats.class);
+    when(aggStats.getStoreStats(STORE_NAME)).thenReturn(storeStats);
+
+    ServerStatsContext statsContext = mock(ServerStatsContext.class);
+    when(statsContext.getResponseStatus()).thenReturn(responseStatus);
+    when(statsContext.getStoreName()).thenReturn(STORE_NAME);
+    when(statsContext.getCurrentStats()).thenReturn(aggStats);
+    when(statsContext.getRequestStartTimeInNS()).thenReturn(0L);
+
+    @SuppressWarnings("unchecked")
+    StreamObserver<VeniceServerResponse> responseObserver = mock(StreamObserver.class);
+    GrpcRequestContext ctx = new GrpcRequestContext(null, VeniceServerResponse.newBuilder(), responseObserver);
+    ctx.setGrpcStatsContext(statsContext);
+    if (hasError) {
+      ctx.setError();
+    }
+
+    new GrpcOutboundStatsHandler().processRequest(ctx);
+
+    if (expectSuccess == null) {
+      verify(statsContext, never()).successRequest(eq(storeStats), anyDouble());
+      verify(statsContext, never()).errorRequest(eq(storeStats), anyDouble());
+    } else if (expectSuccess) {
+      verify(statsContext, times(1)).successRequest(eq(storeStats), anyDouble());
+      verify(statsContext, never()).errorRequest(eq(storeStats), anyDouble());
+    } else {
+      verify(statsContext, times(1)).errorRequest(eq(storeStats), anyDouble());
+      verify(statsContext, never()).successRequest(eq(storeStats), anyDouble());
+    }
+  }
+}