Revert named-blob client-disconnect guard (#3279, #3281)

ambry-frontend/src/main/java/com/github/ambry/frontend/AmbryIdConverterFactory.java

@@ -34,8 +34,6 @@
 import com.github.ambry.rest.RestUtils;
 import com.github.ambry.utils.Pair;
 import com.github.ambry.utils.Utils;
-import java.io.IOException;
-import java.nio.channels.ClosedChannelException;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Objects;
@@ -44,6 +42,7 @@
 import java.util.concurrent.Future;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import com.github.ambry.utils.Utils;
 
 import static com.github.ambry.rest.RestUtils.Headers.*;
 import static com.github.ambry.rest.RestUtils.InternalKeys.*;
@@ -165,7 +164,7 @@ private <T> void completeConversion(T conversionResult, Exception exception, Com
      * @throws RestServiceException
      */
     private CompletionStage<String> convertId(String input, RestRequest restRequest, BlobProperties blobProperties)
-        throws RestServiceException, IOException {
+        throws RestServiceException {
       CompletionStage<String> conversionFuture;
       LOGGER.debug("input for convertId : " + input);
       LOGGER.debug("restRequest for convertId : " + restRequest);
@@ -212,29 +211,6 @@ private CompletionStage<String> convertId(String input, RestRequest restRequest,
           });
         } else {
           Objects.requireNonNull(blobProperties, "blobProperties cannot be null.");
-          // Best-effort: if the client channel has already been closed (e.g. TCP disconnect / stream reset while
-          // router upload was still in flight), skip the metadata commit so the caller's retry can win MAX(version)
-          // in MySqlNamedBlobDb instead of being silently overwritten by this now-orphan attempt. Router chunks
-          // already uploaded will self-expire via existing chunk TTL. See RequestChannelClosed javadoc.
-          if (!restRequest.isOpen()) {
-            frontendMetrics.idConverterClientAbortedCount.inc();
-            LOGGER.info("Client disconnected before namedBlobDb.put for {}; skipping metadata commit",
-                restRequest.getUri());
-            // Must be a client-termination IOException, NOT a RestServiceException.
-            // NettyResponseChannel.getErrorResponse() tests `cause instanceof RestServiceException` BEFORE
-            // Utils.isPossibleClientTermination(cause), so any RestServiceException short-circuits
-            // client-termination detection. RestServiceErrorCode.RequestChannelClosed is not in a 4xx group in
-            // ResponseStatus.getResponseStatus(), so it fell through to InternalServerError -> HTTP 500 and
-            // NettyResponseChannel.InternalServerErrorCount. Because the socket is already dead,
-            // maybeWriteResponseMetadata() never writes that 500, so it inflated 5xx metrics without any
-            // client ever observing it.
-            //
-            // Utils.convertToClientTerminationException() yields the IOException shape that
-            // Utils.isPossibleClientTermination() recognizes, so this lands on HTTP 400 +
-            // NettyResponseChannel.ClientEarlyTerminationCount -- identical to what
-            // NettyMessageProcessor.channelInactive() already emits for this exact disconnect.
-            throw Utils.convertToClientTerminationException(new ClosedChannelException());
-          }
           NamedBlobPath namedBlobPath =
               NamedBlobPath.parse(RestUtils.getRequestPath(restRequest), restRequest.getArgs());
           String blobId = RestUtils.stripSlashAndExtensionFromId(input);

ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java

@@ -241,7 +241,6 @@ public class FrontendMetrics {
   // AmbryIdConverter
   public final Histogram idConverterProcessingTimeInMs;
   public final Histogram idConversionDownstreamCallbackTimeInMs;
-  public final Counter idConverterClientAbortedCount;
 
   // GetPeersHandler
   public final Histogram getPeersProcessingTimeInMs;
@@ -688,8 +687,6 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
         metricRegistry.histogram(MetricRegistry.name(AmbryIdConverterFactory.class, "ProcessingTimeInMs"));
     idConversionDownstreamCallbackTimeInMs =
         metricRegistry.histogram(MetricRegistry.name(AmbryIdConverterFactory.class, "DownstreamCallbackTimeInMs"));
-    idConverterClientAbortedCount =
-        metricRegistry.counter(MetricRegistry.name(AmbryIdConverterFactory.class, "ClientAbortedCount"));
     // GetPeersHandler
     getPeersProcessingTimeInMs =
         metricRegistry.histogram(MetricRegistry.name(GetPeersHandler.class, "ProcessingTimeInMs"));

ambry-frontend/src/test/java/com/github/ambry/frontend/AmbryIdConverterFactoryTest.java

@@ -37,7 +37,6 @@
 import com.github.ambry.utils.Pair;
 import com.github.ambry.utils.TestUtils;
 import com.github.ambry.utils.Utils;
-import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -261,75 +260,6 @@ private void testConversionForNamedBlob(IdConverter idConverter, RestMethod rest
         restRequest.getArgs().containsKey(RestUtils.InternalKeys.NAMED_BLOB_VERSION));
   }
 
-  @Test
-  public void ambryIdConverterNamedBlobPutClientDisconnectTest() throws Exception {
-    // Best-effort race guard: when the client has already disconnected, AmbryIdConverterFactory should NOT
-    // commit named-blob metadata (namedBlobDb.put), and should surface a client-termination exception on
-    // both the future and callback paths, while incrementing idConverterClientAbortedCount.
-    //
-    // The exception shape matters: NettyResponseChannel.getErrorResponse() checks
-    // `cause instanceof RestServiceException` BEFORE Utils.isPossibleClientTermination(cause), so throwing a
-    // RestServiceException here bypasses client-termination detection and gets classified as HTTP 500 /
-    // InternalServerErrorCount rather than HTTP 400 / ClientEarlyTerminationCount.
-    Properties properties = new Properties();
-    VerifiableProperties verifiableProperties = new VerifiableProperties(properties);
-    IdSigningService idSigningService = mock(IdSigningService.class);
-    NamedBlobDb namedBlobDb = mock(NamedBlobDb.class);
-    MetricRegistry metricRegistry = new MetricRegistry();
-    AmbryIdConverterFactory ambryIdConverterFactory =
-        new AmbryIdConverterFactory(verifiableProperties, metricRegistry, idSigningService, namedBlobDb);
-    IdConverter idConverter = ambryIdConverterFactory.getIdConverter();
-    assertNotNull("No IdConverter returned", idConverter);
-    PartitionId partitionId = new MockPartitionId(partition, MockClusterMap.DEFAULT_PARTITION_CLASS);
-    BlobId blobId = new BlobId(BLOB_ID_V6, BlobIdType.NATIVE, dataCenterId, accountId, containerId, partitionId, false,
-        BlobDataType.DATACHUNK);
-
-    // Build a named-blob PUT request the same way ambryIdConverterNamedBlobTest does, then close it BEFORE
-    // convert(...) so restRequest.isOpen() returns false when the write branch is entered.
-    JSONObject requestData = new JSONObject();
-    JSONObject headers = new JSONObject();
-    headers.put(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream");
-    requestData.put(MockRestRequest.REST_METHOD_KEY, RestMethod.PUT.name());
-    requestData.put(MockRestRequest.URI_KEY, NAMED_BLOB_PATH);
-    requestData.put(MockRestRequest.HEADERS_KEY, headers);
-    RestRequest restRequest = new MockRestRequest(requestData, null);
-    restRequest.setArg(RestUtils.InternalKeys.REQUEST_PATH,
-        RequestPath.parse(NAMED_BLOB_PATH, Collections.emptyMap(), Collections.emptyList(), "Ambry-test"));
-    BlobInfo blobInfo = new BlobInfo(new BlobProperties(-1, "service", accountId, containerId, false), new byte[0]);
-    restRequest.close();
-    assertFalse("Test precondition: RestRequest must be closed", restRequest.isOpen());
-
-    String metricName =
-        MetricRegistry.name(AmbryIdConverterFactory.class, "ClientAbortedCount");
-    assertTrue("ClientAbortedCount counter must be registered",
-        metricRegistry.getCounters().containsKey(metricName));
-    long beforeCount = metricRegistry.counter(metricName).getCount();
-
-    IdConversionCallback callback = new IdConversionCallback();
-    try {
-      idConverter.convert(restRequest, blobId.getID(), blobInfo.getBlobProperties(), callback).get(5, TimeUnit.SECONDS);
-      fail("ID conversion should have failed because the client disconnected");
-    } catch (ExecutionException e) {
-      Throwable cause = e.getCause();
-      assertFalse("Guard must not throw RestServiceException (would be classified as HTTP 500) but got " + cause,
-          cause instanceof RestServiceException);
-      assertTrue("Expected IOException (Future) but got " + cause, cause instanceof IOException);
-      assertTrue("Exception must be recognized by Utils.isPossibleClientTermination (Future) but got " + cause,
-          Utils.isPossibleClientTermination(cause));
-    }
-    assertNotNull("Callback exception should be set", callback.exception);
-    assertFalse(
-        "Guard must not throw RestServiceException (would be classified as HTTP 500) but got " + callback.exception,
-        callback.exception instanceof RestServiceException);
-    assertTrue("Expected IOException (Callback) but got " + callback.exception,
-        callback.exception instanceof IOException);
-    assertTrue("Exception must be recognized by Utils.isPossibleClientTermination (Callback) but got "
-        + callback.exception, Utils.isPossibleClientTermination(callback.exception));
-    verify(namedBlobDb, never()).put(any(), any(), any());
-    assertEquals("idConverterClientAbortedCount should have incremented by exactly 1", beforeCount + 1,
-        metricRegistry.counter(metricName).getCount());
-  }
-
   /**
    * Callback implementation for testing {@link IdConverter#convert(RestRequest, String, Callback)}.
    */

ambry-rest/src/main/java/com/github/ambry/rest/NettyMessageProcessor.java

@@ -139,15 +139,6 @@ public void channelInactive(ChannelHandlerContext ctx) {
     nettyMetrics.channelDestructionRate.mark();
     if (request != null && request.isOpen()) {
       logger.error("Request {} was aborted because the channel {} became inactive", request.getUri(), ctx.channel());
-      // Flip channelOpen=false so downstream callbacks that check restRequest.isOpen() observe the disconnect
-      // and can short-circuit best-effort work (e.g. named-blob metadata commit in AmbryIdConverterFactory).
-      // NettyRequest.close() is idempotent via channelOpen.compareAndSet(true,false), so double-close on the
-      // normal-completion path is a no-op.
-      try {
-        request.close();
-      } catch (Exception e) {
-        logger.warn("Exception while closing request {} on channelInactive", request.getUri(), e);
-      }
       onRequestAborted(Utils.convertToClientTerminationException(new ClosedChannelException()));
     } else {
       close();

ambry-rest/src/test/java/com/github/ambry/rest/NettyMessageProcessorTest.java

@@ -490,76 +490,6 @@ public void continueHeaderPutRequestCloseRaceWithoutDelayTest() throws Exception
     compareContent(receivedContent, Collections.singletonList(content));
   }
 
-  /**
-   * Verifies that {@link NettyMessageProcessor#channelInactive} flips {@code NettyRequest.channelOpen} to
-   * false so downstream callbacks that later check {@code restRequest.isOpen()} observe the disconnect.
-   * <p>
-   * Regression: previously {@code channelInactive} only called {@code onRequestAborted(...)}, which routes
-   * an exception into the response channel but does NOT close the {@link NettyRequest}. Best-effort race
-   * guards elsewhere (e.g. named-blob metadata commit in {@code AmbryIdConverterFactory}) depend on
-   * {@code restRequest.isOpen() == false} to short-circuit; without this fix they would still run.
-   *
-   * @throws Exception
-   */
-  @Test
-  public void channelInactiveClosesInflightRequestTest() throws Exception {
-    // Custom handler that captures the RestRequest so we can inspect isOpen() after channelInactive fires.
-    CapturingRestRequestHandler capturingHandler = new CapturingRestRequestHandler();
-    capturingHandler.start();
-    try {
-      NettyMessageProcessor processor =
-          new NettyMessageProcessor(NETTY_METRICS, NETTY_CONFIG, PERFORMANCE_CONFIG, capturingHandler);
-      EmbeddedChannel channel = new EmbeddedChannel(new ChunkedWriteHandler(), processor);
-
-      // Send a PUT header only (no LastHttpContent) so the request stays in-flight when we close the channel.
-      HttpRequest httpRequest = RestTestUtils.createRequest(HttpMethod.PUT, "/", null);
-      httpRequest.headers().set(RestUtils.Headers.SERVICE_ID, "channelInactiveClosesInflightRequestTest");
-      httpRequest.headers().set(RestUtils.Headers.AMBRY_CONTENT_TYPE, "application/octet-stream");
-      channel.writeInbound(httpRequest);
-
-      RestRequest capturedRequest = capturingHandler.getCapturedRequest();
-      assertNotNull("Handler should have received the in-flight RestRequest", capturedRequest);
-      assertTrue("RestRequest must be open before channelInactive", capturedRequest.isOpen());
-
-      // Simulate the client TCP disconnect / channel becoming inactive mid-request.
-      channel.close().awaitUninterruptibly();
-
-      assertFalse("RestRequest.isOpen() must be false after channelInactive so downstream callbacks "
-          + "(e.g. named-blob metadata commit) can observe the disconnect", capturedRequest.isOpen());
-    } finally {
-      capturingHandler.shutdown();
-    }
-  }
-
-  /**
-   * {@link RestRequestHandler} that captures the first {@link RestRequest} passed to
-   * {@link #handleRequest(RestRequest, RestResponseChannel)} and does nothing else. Used by
-   * {@link #channelInactiveClosesInflightRequestTest()} to hold a reference to an in-flight request
-   * so its {@code isOpen()} state can be observed after {@code channelInactive}.
-   */
-  private static class CapturingRestRequestHandler implements RestRequestHandler {
-    private final java.util.concurrent.atomic.AtomicReference<RestRequest> captured =
-        new java.util.concurrent.atomic.AtomicReference<>();
-
-    @Override
-    public void start() {
-    }
-
-    @Override
-    public void shutdown() {
-    }
-
-    @Override
-    public void handleRequest(RestRequest restRequest, RestResponseChannel restResponseChannel) {
-      captured.compareAndSet(null, restRequest);
-      // Do NOT complete the request — leave it in-flight so channelInactive fires before completion.
-    }
-
-    RestRequest getCapturedRequest() {
-      return captured.get();
-    }
-  }
-
   // helpers
   // general