elasticsearchmachine · Jan 21, 2026 · Dec 24, 2025
diff --git a/docs/changelog/138951.yaml b/docs/changelog/138951.yaml
@@ -0,0 +1,6 @@
+pr: 138951
+summary: Retry bulk-delete items in GCS
+area: Snapshot/Restore
+type: enhancement
+issues:
+ - 138364
diff --git a/...t/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java b/...t/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreRepositoryTests.java
@@ -55,7 +55,9 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.IntStream;
 
 import static org.elasticsearch.common.io.Streams.readFully;
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomPurpose;
@@ -124,13 +126,30 @@ protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
         return settings.build();
     }
 
-    public void testDeleteSingleItem() throws IOException {
-        final String repoName = createRepository(randomRepositoryName());
-        final RepositoriesService repositoriesService = internalCluster().getAnyMasterNodeInstance(RepositoriesService.class);
-        final BlobStoreRepository repository = (BlobStoreRepository) repositoriesService.repository(repoName);
-        repository.blobStore()
-            .blobContainer(repository.basePath())
-            .deleteBlobsIgnoringIfNotExists(randomPurpose(), Iterators.single("foo"));
+    public void testDeleteItems() throws IOException {
+        final var repoName = createRepository(randomRepositoryName(), false);
+        final var repositoriesService = internalCluster().getAnyMasterNodeInstance(RepositoriesService.class);
+        final var repository = (BlobStoreRepository) repositoriesService.repository(repoName);
+        final var blobStore = repository.blobStore();
+        final var container = blobStore.blobContainer(repository.basePath());
+
+        final var purpose = randomPurpose();
+        final var blobNamePrefix = "delete-blob-";
+        final int numberOfBlobs = between(1, GoogleCloudStorageBlobStore.MAX_DELETES_PER_BATCH * 10);
+        final List<String> blobNames = IntStream.range(0, numberOfBlobs).mapToObj(n -> blobNamePrefix + n).toList();
+
+        // randomly skips blob creation to exercise deletion if blob not exists
+        int created = 0;
+        for (var blob : blobNames) {
+            if (randomBoolean()) {
+                container.writeBlob(purpose, blob, randomBytesReference(between(1, 10)), false);
+                created += 1;
+            }
+        }
+        assertEquals("should write all blobs", created, container.listBlobsByPrefix(purpose, blobNamePrefix).size());
+
+        container.deleteBlobsIgnoringIfNotExists(purpose, blobNames.iterator());
+        assertEquals("should delete all blobs", 0, container.listBlobsByPrefix(purpose, blobNamePrefix).size());
     }
 
     public void testChunkSize() {

diff --git a/...ory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java b/...ory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GcsRepositoryStatsCollector.java
@@ -190,6 +190,18 @@ public void collectRunnable(OperationPurpose purpose, StorageOperation operation
         }
     }
 
+    /**
+     * Tracks stats, but does not add them to collector. For example DELETE operations are not tracked. It might change in the future.
+     */
+    public void skipCollectRunnable(OperationPurpose purpose, StorageOperation operation, Runnable runnable) {
+        initAndGetThreadLocal(purpose, operation);
+        try {
+            runnable.run();
+        } finally {
+            clearThreadLocal();
+        }
+    }
+
     public <T> T collectIOSupplier(OperationPurpose purpose, StorageOperation operation, IOSupplier<T> blobFn) throws IOException {
         var t = timer.absoluteTimeInMillis();
         var stats = initAndGetThreadLocal(purpose, operation);

diff --git a/...ory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java b/...ory-gcs/src/main/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStore.java
@@ -10,14 +10,13 @@
 package org.elasticsearch.repositories.gcs;
 
 import com.google.cloud.BaseServiceException;
-import com.google.cloud.BatchResult;
 import com.google.cloud.WriteChannel;
 import com.google.cloud.storage.Blob;
 import com.google.cloud.storage.BlobId;
 import com.google.cloud.storage.BlobInfo;
 import com.google.cloud.storage.Storage;
 import com.google.cloud.storage.Storage.BlobListOption;
-import com.google.cloud.storage.StorageBatch;
+import com.google.cloud.storage.StorageBatchResult;
 import com.google.cloud.storage.StorageException;
 
 import org.apache.logging.log4j.LogManager;
@@ -56,18 +55,16 @@
 import java.nio.file.FileAlreadyExistsException;
 import java.util.ArrayList;
 import java.util.Base64;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static java.net.HttpURLConnection.HTTP_GONE;
-import static java.net.HttpURLConnection.HTTP_NOT_FOUND;
 import static java.net.HttpURLConnection.HTTP_PRECON_FAILED;
 import static org.elasticsearch.core.Strings.format;
+import static org.elasticsearch.rest.RestStatus.TOO_MANY_REQUESTS;
 
 class GoogleCloudStorageBlobStore implements BlobStore {
 
@@ -83,7 +80,9 @@ class GoogleCloudStorageBlobStore implements BlobStore {
     // called "resumable upload")
     // https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
     public static final int LARGE_BLOB_THRESHOLD_BYTE_SIZE;
-    public static final int MAX_DELETES_PER_BATCH = 1000;
+
+    // MAX deletes per batch is 100 https://docs.cloud.google.com/storage/docs/batch#overview
+    public static final int MAX_DELETES_PER_BATCH = 100;
 
     static {
         final String key = "es.repository_gcs.large_blob_threshold_byte_size";
@@ -541,72 +540,6 @@ public String next() {
         return deleteResult;
     }
 
-    /**
-     * Deletes multiple blobs from the specific bucket using a batch request
-     *
-     * @param purpose the operation purpose
-     * @param blobNames names of the blobs to delete
-     */
-    void deleteBlobs(OperationPurpose purpose, Iterator<String> blobNames) throws IOException {
-        if (blobNames.hasNext() == false) {
-            return;
-        }
-        final Iterator<BlobId> blobIdsToDelete = new Iterator<>() {
-            @Override
-            public boolean hasNext() {
-                return blobNames.hasNext();
-            }
-
-            @Override
-            public BlobId next() {
-                return BlobId.of(bucketName, blobNames.next());
-            }
-        };
-        final List<BlobId> failedBlobs = Collections.synchronizedList(new ArrayList<>());
-        try {
-            final AtomicReference<StorageException> ioe = new AtomicReference<>();
-            StorageBatch batch = client().batch();
-            int pendingDeletesInBatch = 0;
-            while (blobIdsToDelete.hasNext()) {
-                BlobId blob = blobIdsToDelete.next();
-                batch.delete(blob).notify(new BatchResult.Callback<>() {
-                    @Override
-                    public void success(Boolean result) {}
-
-                    @Override
-                    public void error(StorageException exception) {
-                        if (exception.getCode() != HTTP_NOT_FOUND) {
-                            // track up to 10 failed blob deletions for the exception message below
-                            if (failedBlobs.size() < 10) {
-                                failedBlobs.add(blob);
-                            }
-                            if (ioe.compareAndSet(null, exception) == false) {
-                                ioe.get().addSuppressed(exception);
-                            }
-                        }
-                    }
-                });
-                pendingDeletesInBatch++;
-                if (pendingDeletesInBatch % MAX_DELETES_PER_BATCH == 0) {
-                    batch.submit();
-                    batch = client().batch();
-                    pendingDeletesInBatch = 0;
-                }
-            }
-            if (pendingDeletesInBatch > 0) {
-                batch.submit();
-            }
-
-            final StorageException exception = ioe.get();
-            if (exception != null) {
-                throw exception;
-            }
-        } catch (final Exception e) {
-            throw new IOException("Exception when deleting blobs " + failedBlobs, e);
-        }
-        assert failedBlobs.isEmpty();
-    }
-
     private static String buildKey(String keyPath, String s) {
         assert s != null;
         return keyPath + s;
@@ -739,7 +672,7 @@ OptionalBytesReference compareAndExchangeRegister(
                 if (statusCode == RestStatus.PRECONDITION_FAILED.getStatus()) {
                     return OptionalBytesReference.MISSING;
                 }
-                if (statusCode == RestStatus.TOO_MANY_REQUESTS.getStatus()) {
+                if (statusCode == TOO_MANY_REQUESTS.getStatus()) {
                     finalException = ExceptionsHelper.useOrSuppress(finalException, serviceException);
                     if (retries.hasNext()) {
                         try {
@@ -757,6 +690,105 @@ OptionalBytesReference compareAndExchangeRegister(
         }
     }
 
+    // GCS retry codes https://docs.cloud.google.com/storage/docs/retry-strategy#java
+    private static boolean isRetryErrCode(int code) {
+        final var status = RestStatus.fromCode(code);
+        if (status == null) {
+            return false;
+        }
+        return switch (status) {
+            case REQUEST_TIMEOUT, TOO_MANY_REQUESTS, INTERNAL_SERVER_ERROR, BAD_GATEWAY, SERVICE_UNAVAILABLE, GATEWAY_TIMEOUT -> true;
+            default -> false;
+        };
+    }
+
+    /**
+     * Deletes multiple blobs from the specific bucket using a batch request
+     *
+     * @param blobNames names of the blobs to delete
+     */
+    void deleteBlobs(OperationPurpose purpose, Iterator<String> blobNames) throws IOException {
+        if (blobNames.hasNext() == false) {
+            return;
+        }
+
+        record DeleteResult(BlobId blobId, StorageBatchResult<Boolean> result) {}
+        record DeleteFailure(BlobId blobId, int errCode) {}
+
+        // The following algorithm maximizes the size of every batch by merging retryable items
+        // from the previous batch and new items. When batch results have failed items, we first retry
+        // only a single item using the SDK client's retry strategy (exponential-backoff). Retrying
+        // a single item should provide enough time to back-off from throttling or temporary GCS
+        // failures. Once a single item successfully retries, we proceed with the next batch, combining
+        // the remaining failures and new items.
+        //
+        // Which is roughly looks like this:
+        // - create batch of 100 new items
+        // - submit batch
+        // - receive 100 results, with 10 retryable failures
+        // - retry 1 failure using non-batched delete using SDK retry strategy
+        // - (loop) create batch from 9 remaining failures and 91 new items
+
+        final var batchResults = new ArrayList<DeleteResult>(MAX_DELETES_PER_BATCH);
+        final var batchFailures = new ArrayList<DeleteFailure>(MAX_DELETES_PER_BATCH);
+        while (blobNames.hasNext() || batchFailures.isEmpty() == false) {
+
+            // create a new batch from failed and new items, failed first
+            final var batch = client().batch();
+            for (var failure : batchFailures) {
+                batchResults.add(new DeleteResult(failure.blobId, batch.delete(failure.blobId)));
+            }
+            batchFailures.clear();
+            while (blobNames.hasNext() && batchResults.size() < MAX_DELETES_PER_BATCH) {
+                final var blobId = BlobId.of(bucketName, blobNames.next());
+                batchResults.add(new DeleteResult(blobId, batch.delete(blobId)));
+            }
+
+            // The whole batch request uses GCS client's retry logic, but individual item failures are not retried.
+            // Collect all retryable failures or terminate on non-retryable error.
+            try {
+                batch.submit();
+            } catch (Exception e) {
+                throw new IOException("Failed to execute batch", e);
+            }
+            StorageException nonRetryableException = null;
+            for (var deleteResult : batchResults) {
+                try {
+                    deleteResult.result.get();
+                } catch (StorageException e) {
+                    final var errCode = e.getCode();
+                    batchFailures.add(new DeleteFailure(deleteResult.blobId, e.getCode()));
+                    if (nonRetryableException == null && isRetryErrCode(errCode) == false) {
+                        nonRetryableException = e;
+                    }
+                }
+            }
+            if (nonRetryableException != null) {
+                throw new IOException(
+                    "One or more batch items failed, non-retryable exception; all batch failures: " + batchFailures,
+                    nonRetryableException
+                );
+            }
+            batchResults.clear();
+
+            // Delete single item using GCS client's retry logic.
+            // It should provide enough back-off before trying next batch.
+            if (batchFailures.isEmpty() == false) {
+                final var retryBlobId = batchFailures.getLast().blobId;
+                try {
+                    client().delete(purpose, retryBlobId);
+                    // remaining items go into the next batch
+                    batchFailures.removeLast();
+                } catch (StorageException e) {
+                    throw new IOException(
+                        "Failed to retry single batch item, blobId=" + retryBlobId + "; all batch failures: " + batchFailures,
+                        e
+                    );
+                }
+            }
+        }
+    }
+
     private static BaseServiceException unwrapServiceException(Throwable t) {
         for (int i = 0; i < 10; i++) {
             if (t == null) {

diff --git a/modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/MeteredStorage.java b/modules/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/MeteredStorage.java
@@ -33,6 +33,7 @@
 import java.util.Iterator;
 import java.util.stream.Stream;
 
+import static org.elasticsearch.repositories.gcs.StorageOperation.DELETE;
 import static org.elasticsearch.repositories.gcs.StorageOperation.GET;
 import static org.elasticsearch.repositories.gcs.StorageOperation.INSERT;
 import static org.elasticsearch.repositories.gcs.StorageOperation.LIST;
@@ -82,6 +83,10 @@ public Blob meteredGet(OperationPurpose purpose, BlobId blobId) throws IOExcepti
         return statsCollector.collectIOSupplier(purpose, GET, () -> storage.get(blobId));
     }
 
+    public void delete(OperationPurpose purpose, BlobId blobId) {
+        statsCollector.skipCollectRunnable(purpose, DELETE, () -> storage.delete(blobId));
+    }
+
     public void meteredCreate(
         OperationPurpose purpose,
         BlobInfo blobInfo,

diff --git a/...les/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/StorageOperation.java b/...les/repository-gcs/src/main/java/org/elasticsearch/repositories/gcs/StorageOperation.java
@@ -12,7 +12,8 @@
 public enum StorageOperation {
     INSERT("InsertObject"),
     GET("GetObject"),
-    LIST("ListObjects");
+    LIST("ListObjects"),
+    DELETE("DeleteObject");
 
     final String key;
 

diff --git a/...st/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreContainerTests.java b/...st/java/org/elasticsearch/repositories/gcs/GoogleCloudStorageBlobStoreContainerTests.java
@@ -53,11 +53,7 @@ public void testDeleteBlobsIgnoringIfNotExistsThrowsIOException() throws Excepti
         } else {
             StorageBatchResult<Boolean> resultA = mock(StorageBatchResult.class);
             doReturn(resultA).when(batch).delete(eq(BlobId.of("bucket", "blobA")));
-            doAnswer(invocation -> {
-                StorageException storageException = new StorageException(new IOException("Batched delete throws a storage exception"));
-                ((BatchResult.Callback) invocation.getArguments()[0]).error(storageException);
-                return null;
-            }).when(resultA).notify(any(StorageBatchResult.Callback.class));
+            doThrow(new StorageException(new IOException("Batch item delete throws exception"))).when(resultA).get();
 
             StorageBatchResult<Boolean> resultB = mock(StorageBatchResult.class);
             doReturn(resultB).when(batch).delete(eq(BlobId.of("bucket", "blobB")));