fix(inkless:storage): bound bulk-delete failure logging

docs/inkless/metrics.rst

@@ -236,14 +236,15 @@ FileCleaner metrics
 io.aiven.inkless.delete:type=FileCleaner
 ----------------------------------------
 
-=====================  =========================================================
-Attribute name         Description                                              
-=====================  =========================================================
-FileCleanerErrorRate   Total number of file cleaning errors                     
-FileCleanerFilesRate   Total number of files cleaned                            
-FileCleanerRate        Total number of file cleaning cycles started             
-FileCleanerTotalTime   Total time spent on a file cleaning cycle in milliseconds
-=====================  =========================================================
+===========================  =================================================================================================================================
+Attribute name               Description                                                                                                                      
+===========================  =================================================================================================================================
+FileCleanerErrorRate         Total number of file cleaning errors                                                                                             
+FileCleanerFilesFailedRate   Total number of files the storage backend did not confirm deleted; they stay marked for deletion and are retried on a later cycle
+FileCleanerFilesRate         Total number of files cleaned                                                                                                    
+FileCleanerRate              Total number of file cleaning cycles started                                                                                     
+FileCleanerTotalTime         Total time spent on a file cleaning cycle in milliseconds                                                                        
+===========================  =================================================================================================================================
 
 
 RetentionEnforcer metrics

storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java

@@ -108,22 +108,10 @@ public void run() {
             } else {
                 LOGGER.info("Running file cleaner: deleting {} of {} marked files", objectKeyPaths.size(), filesToDelete.size());
                 metrics.recordFileCleanerStart();
-                // 1-element holder to carry the duration out of the (synchronous, same-thread) callback
-                // for the log line below; a plain local cannot be assigned from the lambda.
-                final long[] durationMs = {0};
-                TimeUtils.measureDurationMs(time, () -> {
-                    try {
-                        cleanFiles(objectKeyPaths);
-                    } catch (StorageBackendException e) {
-                        LOGGER.error("Error while cleaning files", e);
-                        throw new RuntimeException(e);
-                    }
-                }, duration -> {
-                    durationMs[0] = duration;
-                    metrics.recordFileCleanerTotalTime(duration);
-                });
-                metrics.recordFileCleanerCompleted(objectKeyPaths.size());
-                LOGGER.info("File cleaner deleted {} files in {} ms", objectKeyPaths.size(), durationMs[0]);
+                final int deletedCount = TimeUtils.measureDurationMs(time,
+                    () -> cleanFiles(objectKeyPaths),
+                    metrics::recordFileCleanerTotalTime);
+                LOGGER.info("File cleaner deleted {} of {} files", deletedCount, objectKeyPaths.size());
             }
 
             attempts.set(0);
@@ -135,15 +123,30 @@ public void run() {
         }
     }
 
-    private void cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
+    private int cleanFiles(Set<String> objectKeyPaths) throws StorageBackendException {
         final Set<ObjectKey> objectKeys = objectKeyPaths.stream()
             .map(objectKeyCreator::from)
             .collect(Collectors.toSet());
-        // delete files from storage backend
-        storage.delete(objectKeys);
+        // Delete files from the storage backend. Deletion may be partial (e.g. under S3 throttling):
+        // only the keys the backend confirmed deleted are dereferenced in the control plane, so the
+        // remaining keys stay marked for deletion and are retried on the next cycle instead of being
+        // re-attempted after already being deleted.
+        final Set<ObjectKey> deletedKeys = storage.delete(objectKeys);
+        metrics.recordFileCleanerFilesFailed(objectKeyPaths.size() - deletedKeys.size());
+        if (deletedKeys.isEmpty()) {
+            LOGGER.warn("No files deleted from storage out of {} candidates; retrying next cycle",
+                objectKeyPaths.size());
+            return 0;
+        }
+        final Set<String> deletedPaths = deletedKeys.stream()
+            .map(ObjectKey::value)
+            .collect(Collectors.toSet());
         // update control plane
-        final DeleteFilesRequest request = new DeleteFilesRequest(objectKeyPaths);
+        final DeleteFilesRequest request = new DeleteFilesRequest(deletedPaths);
         controlPlane.deleteFiles(request);
+
+        metrics.recordFileCleanerCompleted(deletedPaths.size());
+        return deletedPaths.size();
     }
 
     @Override

storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleanerMetrics.java

@@ -37,6 +37,9 @@ public class FileCleanerMetrics {
     private static final String FILE_CLEANER_FILES_RATE_DOC = "Total number of files cleaned";
     static final String FILE_CLEANER_ERROR_RATE = "FileCleanerErrorRate";
     private static final String FILE_CLEANER_ERROR_RATE_DOC = "Total number of file cleaning errors";
+    static final String FILE_CLEANER_FILES_FAILED_RATE = "FileCleanerFilesFailedRate";
+    private static final String FILE_CLEANER_FILES_FAILED_RATE_DOC = "Total number of files the storage backend did "
+        + "not confirm deleted; they stay marked for deletion and are retried on a later cycle";
 
     /**
      * This method returns a list of all the metric name templates for the FileCleanerMetrics class.
@@ -47,7 +50,8 @@ public static List<MetricNameTemplate> all() {
             new MetricNameTemplate(FILE_CLEANER_TOTAL_TIME, GROUP, FILE_CLEANER_TOTAL_TIME_DOC),
             new MetricNameTemplate(FILE_CLEANER_RATE, GROUP, FILE_CLEANER_RATE_DOC),
             new MetricNameTemplate(FILE_CLEANER_FILES_RATE, GROUP, FILE_CLEANER_FILES_RATE_DOC),
-            new MetricNameTemplate(FILE_CLEANER_ERROR_RATE, GROUP, FILE_CLEANER_ERROR_RATE_DOC)
+            new MetricNameTemplate(FILE_CLEANER_ERROR_RATE, GROUP, FILE_CLEANER_ERROR_RATE_DOC),
+            new MetricNameTemplate(FILE_CLEANER_FILES_FAILED_RATE, GROUP, FILE_CLEANER_FILES_FAILED_RATE_DOC)
         );
     }
 
@@ -57,12 +61,15 @@ public static List<MetricNameTemplate> all() {
     private final LongAdder fileCleanerRate = new LongAdder();
     private final LongAdder fileCleanerFiles = new LongAdder();
     private final LongAdder fileCleanerErrorRate = new LongAdder();
+    // package-private for tests, following ClientAzAwarenessMetrics
+    final LongAdder fileCleanerFilesFailed = new LongAdder();
 
     public FileCleanerMetrics() {
         fileCleanerTotalTime = metricsGroup.newHistogram(FILE_CLEANER_TOTAL_TIME, true, Map.of());
         metricsGroup.newGauge(FILE_CLEANER_RATE, fileCleanerRate::intValue);
         metricsGroup.newGauge(FILE_CLEANER_FILES_RATE, fileCleanerFiles::intValue);
         metricsGroup.newGauge(FILE_CLEANER_ERROR_RATE, fileCleanerErrorRate::intValue);
+        metricsGroup.newGauge(FILE_CLEANER_FILES_FAILED_RATE, fileCleanerFilesFailed::intValue);
     }
 
     public void recordFileCleanerStart() {
@@ -81,10 +88,15 @@ public void recordFileCleanerCompleted(int filesSize) {
         fileCleanerFiles.add(filesSize);
     }
 
+    public void recordFileCleanerFilesFailed(int filesSize) {
+        fileCleanerFilesFailed.add(filesSize);
+    }
+
     public void close() {
         metricsGroup.removeMetric(FILE_CLEANER_TOTAL_TIME);
         metricsGroup.removeMetric(FILE_CLEANER_RATE);
         metricsGroup.removeMetric(FILE_CLEANER_FILES_RATE);
         metricsGroup.removeMetric(FILE_CLEANER_ERROR_RATE);
+        metricsGroup.removeMetric(FILE_CLEANER_FILES_FAILED_RATE);
     }
 }

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java

@@ -32,18 +32,23 @@
 import com.azure.storage.common.StorageSharedKeyCredential;
 import com.groupcdg.pitest.annotations.CoverageIgnore;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.BufferedOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.channels.Channels;
 import java.nio.channels.ReadableByteChannel;
+import java.util.HashSet;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
 import io.aiven.inkless.common.ByteRange;
 import io.aiven.inkless.common.ObjectKey;
+import io.aiven.inkless.storage_backend.common.DeleteErrorSummary;
 import io.aiven.inkless.storage_backend.common.InvalidRangeException;
 import io.aiven.inkless.storage_backend.common.KeyNotFoundException;
 import io.aiven.inkless.storage_backend.common.StorageBackend;
@@ -52,6 +57,14 @@
 
 @CoverageIgnore // tested on integration level
 public final class AzureBlobStorage extends StorageBackend {
+    private static final Logger LOGGER = LoggerFactory.getLogger(AzureBlobStorage.class);
+
+    // Azure error codes that indicate throttling or a transient server condition rather than a hard,
+    // non-transient failure. Used only to log throttling distinctly; both kinds are left for the next
+    // FileCleaner cycle to retry.
+    private static final Set<String> THROTTLE_ERROR_CODES =
+        Set.of("ServerBusy", "InternalError", "OperationTimedOut");
+
     private AzureBlobStorageConfig config;
     private BlobContainerClient blobContainerClient;
     private MetricCollector.MetricsPolicy policy;
@@ -195,16 +208,28 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
-        try {
-            for (ObjectKey key : keys) {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
+        // Deleting one blob at a time (there is no Azure batch-delete dependency here), so a failure
+        // on one key must not abandon the rest: accumulate the keys that were removed and report the
+        // failed ones as not deleted. deleteIfExists() returns true if the blob was deleted and false
+        // if it was already absent; both mean the key is gone (idempotent).
+        final Set<ObjectKey> deleted = new HashSet<>();
+        final DeleteErrorSummary errors = new DeleteErrorSummary();
+        for (final ObjectKey key : keys) {
+            try {
                 blobContainerClient.getBlobClient(key.value()).deleteIfExists();
+                deleted.add(key);
+            } catch (final BlobStorageException e) {
+                final String code = String.valueOf(e.getErrorCode());
+                errors.record(code, key + ": " + e.getMessage(), THROTTLE_ERROR_CODES.contains(code), e);
+            } catch (final RuntimeException e) {
+                final Throwable unwrapped = Exceptions.unwrap(e);
+                errors.record(unwrapped.getClass().getSimpleName(),
+                    key + ": " + unwrapped.getMessage(), false, unwrapped);
             }
-        } catch (final BlobStorageException e) {
-            throw new StorageBackendException("Failed to delete " + keys, e);
-        } catch (final RuntimeException e) {
-            throw unwrapReactorExceptions(e, "Failed to delete " + keys);
         }
+        errors.log(LOGGER, keys.size(), deleted.size());
+        return deleted;
     }
 
     private StorageBackendException unwrapReactorExceptions(final RuntimeException e, final String message) {

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/DeleteErrorSummary.java

@@ -0,0 +1,101 @@
+/*
+ * Inkless
+ * Copyright (C) 2024 - 2026 Aiven OY
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package io.aiven.inkless.storage_backend.common;
+
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeMap;
+
+/**
+ * Accumulates the per-key failures of one bulk delete into a single log line, bounded by the number of
+ * distinct error codes rather than by the number of keys submitted.
+ *
+ * <p>Throttled keys are counted but never sampled: they carry no per-key information and dominate the
+ * failures under load. A call that failed only from throttling is reported at INFO, since it is normal
+ * backpressure rather than an operator-actionable failure; anything else is reported at WARN.
+ */
+public final class DeleteErrorSummary {
+
+    private static final int MAX_SAMPLES = 3;
+
+    private final Map<String, Integer> countsByCode = new TreeMap<>();
+    private final List<String> samples = new ArrayList<>();
+    private Throwable firstHardFailure;
+
+    public void record(final String errorCode, final String keyDetail, final boolean throttled) {
+        record(errorCode, keyDetail, throttled, null);
+    }
+
+    /**
+     * @param cause reported with the summary line for the first hard failure only, so that one stack
+     *              trace per delete call reaches the log instead of one per key.
+     */
+    public void record(final String errorCode,
+                       final String keyDetail,
+                       final boolean throttled,
+                       final Throwable cause) {
+        countsByCode.merge(errorCode, 1, Integer::sum);
+        if (throttled) {
+            return;
+        }
+        if (samples.size() < MAX_SAMPLES) {
+            samples.add(keyDetail);
+        }
+        if (firstHardFailure == null) {
+            firstHardFailure = cause;
+        }
+    }
+
+    public boolean isEmpty() {
+        return countsByCode.isEmpty();
+    }
+
+    public void log(final Logger logger, final int requested, final int deletedCount) {
+        if (isEmpty()) {
+            return;
+        }
+        final String summary = format(requested, deletedCount);
+        if (samples.isEmpty()) {
+            logger.info(summary);
+        } else if (firstHardFailure != null) {
+            logger.warn(summary, firstHardFailure);
+        } else {
+            logger.warn(summary);
+        }
+    }
+
+    // visible for testing
+    String format(final int requested, final int deletedCount) {
+        return String.format(
+            "Failed to delete %d of %d keys %s%s",
+            requested - deletedCount,
+            requested,
+            countsByCode,
+            samples.isEmpty() ? "" : "; e.g. " + String.join("; ", samples)
+        );
+    }
+
+    // visible for testing
+    Optional<Throwable> firstHardFailure() {
+        return Optional.ofNullable(firstHardFailure);
+    }
+}

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java

@@ -34,6 +34,13 @@ public interface ObjectDeleter extends Closeable {
      * Delete objects from a set of keys.
      *
      * <p>If the object doesn't exist, the operation still succeeds as it is idempotent.
+     *
+     * <p>Deletion may be partial: implementations return the subset of {@code keys} that were
+     * confirmed deleted (which includes keys that were already absent). Keys omitted from the
+     * returned set were not deleted this round (e.g. throttled) and are safe to retry, since
+     * deletion is idempotent. Implementations may still throw for a total/unexpected failure.
+     *
+     * @return the subset of {@code keys} confirmed deleted.
      */
-    void delete(Set<ObjectKey> keys) throws StorageBackendException;
+    Set<ObjectKey> delete(Set<ObjectKey> keys) throws StorageBackendException;
 }

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java

@@ -124,15 +124,21 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
         try {
             final Set<BlobId> ids = keys.stream()
                     .map(k -> BlobId.of(this.bucketName,k.value()))
                     .collect(Collectors.toSet());
 
+            // storage.delete returns a List<Boolean> of deleted-vs-already-absent, but a genuine
+            // failure surfaces as a thrown BaseServiceException rather than a per-blob flag, so we
+            // cannot extract a confirmed-deleted subset the way the S3 backend does. This stays
+            // all-or-nothing: on success every key is gone (idempotent), and on failure we delete
+            // nothing and let the FileCleaner cycle retry the whole set.
             storage.delete(ids);
+            return Set.copyOf(keys);
         } catch (final BaseServiceException e) {
-            throw new StorageBackendException("Failed to delete " + keys, e);
+            throw new StorageBackendException("Failed to delete " + keys.size() + " keys", e);
         }
     }
 

storage/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java

@@ -110,9 +110,10 @@ public void delete(final ObjectKey key) throws StorageBackendException {
     }
 
     @Override
-    public void delete(final Set<ObjectKey> keys) throws StorageBackendException {
+    public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
         Objects.requireNonNull(keys, "keys cannot be null");
         keys.forEach(storage::remove);
+        return Set.copyOf(keys);
     }
 
     @Override