address review comments

Author: pxsalehi

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobStore.java

@@ -52,7 +52,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Collectors;
 
@@ -342,7 +341,17 @@ public BlobContainer blobContainer(BlobPath path) {
         return new S3BlobContainer(path, this);
     }
 
-    record DeletionExceptions(Exception exception, int count) {}
+    private static class DeletionExceptions {
+        private Exception exception = null;
+        private int count = 0;
+
+        void useOrMaybeSuppress(Exception e) {
+            if (count < MAX_DELETE_EXCEPTIONS) {
+                exception = ExceptionsHelper.useOrSuppress(exception, e);
+                count++;
+            }
+        }
+    }
 
     void deleteBlobs(OperationPurpose purpose, Iterator<String> blobNames) throws IOException {
         if (blobNames.hasNext() == false) {
@@ -352,19 +361,19 @@ void deleteBlobs(OperationPurpose purpose, Iterator<String> blobNames) throws IO
         final List<String> partition = new ArrayList<>();
         try {
             // S3 API only allows 1k blobs per delete so we split up the given blobs into requests of max. 1k deletes
-            final AtomicReference<DeletionExceptions> aex = new AtomicReference<>(new DeletionExceptions(null, 0));
+            final var deletionExceptions = new DeletionExceptions();
             blobNames.forEachRemaining(key -> {
                 partition.add(key);
                 if (partition.size() == bulkDeletionBatchSize) {
-                    deletePartition(purpose, partition, aex);
+                    deletePartition(purpose, partition, deletionExceptions);
                     partition.clear();
                 }
             });
             if (partition.isEmpty() == false) {
-                deletePartition(purpose, partition, aex);
+                deletePartition(purpose, partition, deletionExceptions);
             }
-            if (aex.get().exception != null) {
-                throw aex.get().exception;
+            if (deletionExceptions.exception != null) {
+                throw deletionExceptions.exception;
             }
         } catch (Exception e) {
             throw new IOException("Failed to delete blobs " + partition.stream().limit(10).toList(), e);
@@ -376,9 +385,9 @@ void deleteBlobs(OperationPurpose purpose, Iterator<String> blobNames) throws IO
      *
      * @param purpose The {@link OperationPurpose} of the deletion
      * @param partition The list of blobs to delete
-     * @param aex A holder for any exception(s) thrown during the deletion
+     * @param deletionExceptions A holder for any exception(s) thrown during the deletion
      */
-    private void deletePartition(OperationPurpose purpose, List<String> partition, AtomicReference<DeletionExceptions> aex) {
+    private void deletePartition(OperationPurpose purpose, List<String> partition, DeletionExceptions deletionExceptions) {
         final Iterator<TimeValue> retries = retryThrottledDeleteBackoffPolicy.iterator();
         int retryCounter = 0;
         while (true) {
@@ -399,7 +408,7 @@ private void deletePartition(OperationPurpose purpose, List<String> partition, A
                     ),
                     e
                 );
-                useOrMaybeSuppress(aex, e);
+                deletionExceptions.useOrMaybeSuppress(e);
                 return;
             } catch (AmazonClientException e) {
                 if (shouldRetryDelete(purpose) && RetryUtils.isThrottlingException(e)) {
@@ -408,26 +417,19 @@ private void deletePartition(OperationPurpose purpose, List<String> partition, A
                         retryCounter++;
                     } else {
                         s3RepositoriesMetrics.retryDeletesHistogram().record(retryCounter);
-                        useOrMaybeSuppress(aex, e);
+                        deletionExceptions.useOrMaybeSuppress(e);
                         return;
                     }
                 } else {
                     // The AWS client threw any unexpected exception and did not execute the request at all so we do not
                     // remove any keys from the outstanding deletes set.
-                    useOrMaybeSuppress(aex, e);
+                    deletionExceptions.useOrMaybeSuppress(e);
                     return;
                 }
             }
         }
     }
 
-    private void useOrMaybeSuppress(AtomicReference<DeletionExceptions> aex, Exception e) {
-        var deletionExceptions = aex.get();
-        if (deletionExceptions.count < MAX_DELETE_EXCEPTIONS) {
-            aex.set(new DeletionExceptions(ExceptionsHelper.useOrSuppress(deletionExceptions.exception, e), deletionExceptions.count + 1));
-        }
-    }
-
     /**
      * If there are remaining retries, pause for the configured interval then return true
      *

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java

@@ -1105,14 +1105,14 @@ public void testSuppressedDeletionErrorsAreCapped() {
                 fail("expected only deletions");
             }
         });
-        try {
-            var maxNoOfDeletions = 2 * S3BlobStore.MAX_DELETE_EXCEPTIONS;
-            var blobs = randomList(1, maxNoOfDeletions * maxBulkDeleteSize, ESTestCase::randomIdentifier);
-            blobContainer.deleteBlobsIgnoringIfNotExists(randomPurpose(), blobs.iterator());
-            fail("deletion should not succeed");
-        } catch (IOException e) {
-            assertThat(e.getCause().getSuppressed().length, lessThan(S3BlobStore.MAX_DELETE_EXCEPTIONS));
-        }
+        var maxNoOfDeletions = 2 * S3BlobStore.MAX_DELETE_EXCEPTIONS;
+        var blobs = randomList(1, maxNoOfDeletions * maxBulkDeleteSize, ESTestCase::randomIdentifier);
+        var exception = expectThrows(
+            IOException.class,
+            "deletion should not succeed",
+            () -> blobContainer.deleteBlobsIgnoringIfNotExists(randomPurpose(), blobs.iterator())
+        );
+        assertThat(exception.getCause().getSuppressed().length, lessThan(S3BlobStore.MAX_DELETE_EXCEPTIONS));
     }
 
     @Override