S3 batch delete: log per-object failures and add a failure counter (#3257)

S3 DeleteObjects always returns HTTP 200 with per-object errors in the
XML response body. Today, when a sub-op fails inside a batch, the
failure is captured into the response payload but is never logged or
counted on the frontend, so partial failures are invisible to operators
unless every response body is parsed offline.

Two changes:

  - Log every per-object failure at WARN with key, error code, the
    parent request URI, and the exception (stack trace included via
    SLF4J's trailing-throwable convention). Operators can now grep
    'S3 batch delete sub-op failed' in ambry-frontend.log to find
    failed deletes inside otherwise-200 batches.

  - Add s3BatchDeleteSubOpFailureCount Counter to FrontendMetrics.
    Increments per failed sub-op. Lets dashboards/alerts track
    partial-failure rate without log scraping.

No protocol or response-body change. The handler's HTTP behavior is
unchanged: still returns 200 with the same DeleteResult XML containing
deleted/error lists. Existing S3BatchDeleteHandlerTest (6 tests) passes.

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

Author: satishkotha

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

@@ -175,6 +175,11 @@ public class FrontendMetrics {
   public final AsyncOperationTracker.Metrics s3PutHandleMetrics;
   public final AsyncOperationTracker.Metrics s3GetHandleMetrics;
 
+  // Counts every per-object delete failure inside an S3 batch-delete request. Surfaces partial
+  // failures that S3 DeleteObjects returns inside the (HTTP 200) response body, which were
+  // previously invisible to operators without parsing every response payload.
+  public final Counter s3BatchDeleteSubOpFailureCount;
+
   // Rates
   // AmbrySecurityService
   public final Meter securityServicePreProcessRequestRate;
@@ -533,6 +538,8 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon
     s3DeleteHandleMetrics = new AsyncOperationTracker.Metrics(S3DeleteHandler.class, "S3Handle", metricRegistry);
     s3BatchDeleteHandleMetrics =
         new AsyncOperationTracker.Metrics(S3BatchDeleteHandler.class, "S3Handle", metricRegistry);
+    s3BatchDeleteSubOpFailureCount =
+        metricRegistry.counter(MetricRegistry.name(S3BatchDeleteHandler.class, "SubOpFailureCount"));
     s3ListHandleMetrics = new AsyncOperationTracker.Metrics(S3ListHandler.class, "S3Handle", metricRegistry);
     s3PutHandleMetrics = new AsyncOperationTracker.Metrics(S3PutHandler.class, "S3Handle", metricRegistry);
     s3GetHandleMetrics = new AsyncOperationTracker.Metrics(S3GetHandler.class, "S3Handle", metricRegistry);

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

@@ -150,14 +150,16 @@ private Callback<Long> parseRequestBodyAndDeleteCallback(RetainingAsyncWritableC
 
         // Handle the delete operation using the deleteBlobHandler
         deleteBlobHandler.handle(singleDeleteRequest, noOpResponseChannel, (result, exception) -> {
-          // Call our custom onDeleteCompletion to track success/failure
           if (exception == null) {
             deleted.add(new S3MessagePayload.S3DeletedObject(object.getKey()));
-          } else if (exception instanceof RestServiceException) {
-            RestServiceException restServiceException = (RestServiceException) exception;
-            errors.add((new S3MessagePayload.S3ErrorObject(object.getKey(), restServiceException.getErrorCode().toString())));
           } else {
-            errors.add((new S3MessagePayload.S3ErrorObject(object.getKey(), RestServiceErrorCode.InternalServerError.toString())));
+            String errorCode = (exception instanceof RestServiceException)
+                ? ((RestServiceException) exception).getErrorCode().toString()
+                : RestServiceErrorCode.InternalServerError.toString();
+            errors.add(new S3MessagePayload.S3ErrorObject(object.getKey(), errorCode));
+            metrics.s3BatchDeleteSubOpFailureCount.inc();
+            logger.warn("S3 batch delete sub-op failed: key={} errorCode={} requestUri={}",
+                object.getKey(), errorCode, restRequest.getUri(), exception);
           }
           future.complete(null);
         });