Log and meter S3 batch-delete request-body parse failures (#3255)

The catch in S3BatchDeleteHandler.parseRequestBodyAndDeleteCallback
mapped every Exception to a generic MalformedRequestBody S3 error
without logging the cause or incrementing any metric. A genuine
server-side bug (NPE, IOException, etc.) was indistinguishable from a
client sending malformed XML.

Add FrontendMetrics.s3BatchDeleteRequestParseError counter, increment it
in the catch, and log the exception at WARN. The S3 protocol response
(200 OK with XML error body) is unchanged.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: snalli

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

@@ -265,6 +265,8 @@ public class FrontendMetrics {
   // GetReplicasHandler
   public final Counter invalidBlobIdError;
   public final Counter responseConstructionError;
+  // S3BatchDeleteHandler
+  public final Counter s3BatchDeleteRequestParseError;
   // Other
   // FrontendRestRequestService
   public final Histogram restRequestServiceStartupTimeInMs;
@@ -721,6 +723,9 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
     invalidBlobIdError = metricRegistry.counter(MetricRegistry.name(GetReplicasHandler.class, "InvalidBlobIdError"));
     responseConstructionError =
         metricRegistry.counter(MetricRegistry.name(GetReplicasHandler.class, "ResponseConstructionError"));
+    // S3BatchDeleteHandler
+    s3BatchDeleteRequestParseError =
+        metricRegistry.counter(MetricRegistry.name(S3BatchDeleteHandler.class, "RequestParseError"));
 
     // Other
     restRequestServiceStartupTimeInMs =

ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3BatchDeleteHandler.java

@@ -116,6 +116,8 @@ private Callback<Long> parseRequestBodyAndDeleteCallback(RetainingAsyncWritableC
           throw new RestServiceException(batchSizeErrorMessage, RestServiceErrorCode.BadRequest);
         }
       } catch (Exception e) {
+        metrics.s3BatchDeleteRequestParseError.inc();
+        logger.warn("Failed to parse S3 batch-delete request body, returning malformed-body error", e);
         S3MessagePayload.Error response = new S3MessagePayload.Error();
         response.setCode(ERR_MALFORMED_REQUEST_BODY_CODE);
         response.setMessage(ERR_MALFORMED_REQUEST_BODY_MESSAGE);

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

@@ -71,6 +71,7 @@ public class S3BatchDeleteHandlerTest {
   private final Account account;
   private final Container container;
   private FrontendConfig frontendConfig;
+  private FrontendMetrics metrics;
   private NamedBlobPutHandler namedBlobPutHandler;
   private S3BatchDeleteHandler s3BatchDeleteHandler;
   private final NettyByteBufLeakHelper nettyByteBufLeakHelper = new NettyByteBufLeakHelper();
@@ -233,6 +234,7 @@ public void malformedXMLRequestTest() throws Exception {
     RestResponseChannel restResponseChannel = new MockRestResponseChannel();
     request.setArg(RestUtils.InternalKeys.REQUEST_PATH,
         RequestPath.parse(request, frontendConfig.pathPrefixesToRemove, CLUSTER_NAME));
+    long parseErrorsBefore = metrics.s3BatchDeleteRequestParseError.getCount();
     FutureResult<ReadableStreamChannel> futureResult = new FutureResult<>();
     s3BatchDeleteHandler.handle(request, restResponseChannel, futureResult::done);
     ReadableStreamChannel readableStreamChannel = futureResult.get();
@@ -245,6 +247,8 @@ public void malformedXMLRequestTest() throws Exception {
     assertEquals(response.getMessage(), ERR_MALFORMED_REQUEST_BODY_MESSAGE);
     assertEquals(response.getCode(), ERR_MALFORMED_REQUEST_BODY_CODE);
     assertEquals("Mismatch on status", ResponseStatus.Ok, restResponseChannel.getStatus());
+    assertEquals("Parse-error counter should increment by 1", parseErrorsBefore + 1,
+        metrics.s3BatchDeleteRequestParseError.getCount());
   }
 
   @Test
@@ -314,7 +318,7 @@ private void setup() throws Exception {
     CommonTestUtils.populateRequiredRouterProps(properties);
     VerifiableProperties verifiableProperties = new VerifiableProperties(properties);
     frontendConfig = new FrontendConfig(verifiableProperties);
-    FrontendMetrics metrics = new FrontendMetrics(new MetricRegistry(), frontendConfig);
+    metrics = new FrontendMetrics(new MetricRegistry(), frontendConfig);
     AccountAndContainerInjector injector = new AccountAndContainerInjector(ACCOUNT_SERVICE, metrics, frontendConfig);
     IdSigningService idSigningService = new AmbryIdSigningService();
     AmbrySecurityServiceFactory securityServiceFactory =