diff --git a/docs/inkless/metrics.rst b/docs/inkless/metrics.rst index 03538d89f89..379a92cf400 100644 --- a/docs/inkless/metrics.rst +++ b/docs/inkless/metrics.rst @@ -218,14 +218,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 diff --git a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java b/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java index 95b9da71738..e8124ba9a8a 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java +++ b/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 objectKeyPaths) throws StorageBackendException { + private int cleanFiles(Set objectKeyPaths) throws StorageBackendException { final Set 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 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 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 diff --git a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleanerMetrics.java b/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleanerMetrics.java index ac2be98aa29..bea5d8ead6d 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleanerMetrics.java +++ b/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 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 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); } } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java index 729aea6f5f8..7b1a955d491 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java @@ -32,12 +32,16 @@ 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; @@ -52,6 +56,8 @@ @CoverageIgnore // tested on integration level public final class AzureBlobStorage extends StorageBackend { + private static final Logger LOGGER = LoggerFactory.getLogger(AzureBlobStorage.class); + private AzureBlobStorageConfig config; private BlobContainerClient blobContainerClient; private MetricCollector.MetricsPolicy policy; @@ -195,16 +201,23 @@ public void delete(final ObjectKey key) throws StorageBackendException { } @Override - public void delete(final Set keys) throws StorageBackendException { - try { - for (ObjectKey key : keys) { + public Set delete(final Set 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 deleted = new HashSet<>(); + for (final ObjectKey key : keys) { + try { blobContainerClient.getBlobClient(key.value()).deleteIfExists(); + deleted.add(key); + } catch (final BlobStorageException e) { + LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, e); + } catch (final RuntimeException e) { + LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, Exceptions.unwrap(e)); } - } catch (final BlobStorageException e) { - throw new StorageBackendException("Failed to delete " + keys, e); - } catch (final RuntimeException e) { - throw unwrapReactorExceptions(e, "Failed to delete " + keys); } + return deleted; } private StorageBackendException unwrapReactorExceptions(final RuntimeException e, final String message) { diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java index 0ae5b5d742f..41e0bf317fd 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/common/ObjectDeleter.java +++ b/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. * *

If the object doesn't exist, the operation still succeeds as it is idempotent. + * + *

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 keys) throws StorageBackendException; + Set delete(Set keys) throws StorageBackendException; } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java index 63e419229e4..70a369c5fc1 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java @@ -124,13 +124,19 @@ public void delete(final ObjectKey key) throws StorageBackendException { } @Override - public void delete(final Set keys) throws StorageBackendException { + public Set delete(final Set keys) throws StorageBackendException { try { final Set ids = keys.stream() .map(k -> BlobId.of(this.bucketName,k.value())) .collect(Collectors.toSet()); + // storage.delete returns a List 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); } diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java index 577fce56e62..a4d13fddc24 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/in_memory/InMemoryStorage.java +++ b/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 keys) throws StorageBackendException { + public Set delete(final Set keys) throws StorageBackendException { Objects.requireNonNull(keys, "keys cannot be null"); keys.forEach(storage::remove); + return Set.copyOf(keys); } @Override diff --git a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java index 1faf0851732..996ea679934 100644 --- a/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java +++ b/storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java @@ -22,11 +22,15 @@ import com.groupcdg.pitest.annotations.CoverageIgnore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.IOException; import java.io.InputStream; import java.nio.channels.Channels; import java.nio.channels.ReadableByteChannel; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -54,11 +58,20 @@ import software.amazon.awssdk.services.s3.model.GetObjectRequest; import software.amazon.awssdk.services.s3.model.ObjectIdentifier; import software.amazon.awssdk.services.s3.model.PutObjectRequest; +import software.amazon.awssdk.services.s3.model.S3Error; @CoverageIgnore // tested on integration level public final class S3Storage extends StorageBackend { + private static final Logger LOGGER = LoggerFactory.getLogger(S3Storage.class); + public static final int MAX_DELETE_KEYS_LIMIT = 1000; + + // Per-key S3 error codes that indicate throttling rather than a hard, non-transient failure. Used + // only to log throttling distinctly; neither kind is retried in-call. + private static final Set THROTTLE_ERROR_CODES = + Set.of("SlowDown", "ServiceUnavailable", "RequestLimitExceeded"); + private S3Client s3Client; private String bucketName; @@ -155,38 +168,70 @@ public void delete(final ObjectKey key) throws StorageBackendException { } @Override - public void delete(final Set keys) throws StorageBackendException { + public Set delete(final Set keys) throws StorageBackendException { final List objectKeys = new ArrayList<>(keys); - List batch = null; - try { - for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) { - batch = objectKeys.subList( - i, - Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size()) - ); - - final Set ids = batch.stream() - .map(k -> ObjectIdentifier.builder().key(k.value()).build()) - .collect(Collectors.toSet()); - final Delete delete = Delete.builder().objects(ids).build(); - final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder() - .bucket(bucketName) - .delete(delete) - .build(); - final DeleteObjectsResponse response = s3Client.deleteObjects(deleteObjectsRequest); - - if (!response.errors().isEmpty()) { - final var errors = response.errors().stream() - .map(e -> String.format("Error %s: %s (%s)", e.key(), e.message(), e.code())) - .collect(Collectors.joining(", ")); - throw new StorageBackendException("Failed to delete keys " + batch + ": " + errors); + final Set deleted = new HashSet<>(); + for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) { + final Set batch = new HashSet<>(objectKeys.subList( + i, + Math.min(i + MAX_DELETE_KEYS_LIMIT, objectKeys.size()) + )); + final Map byValue = batch.stream() + .collect(Collectors.toMap(ObjectKey::value, k -> k, (a, b) -> a)); + final DeleteObjectsResponse response; + try { + response = deleteObjectsOnce(batch); + } catch (final SdkException e) { + // Whole-request failure, including a 503 the SDK's adaptive retry already exhausted and + // timeouts. Stop this pass and report the remaining keys as not deleted; deletion is + // idempotent, so re-attempting them later is safe. + LOGGER.warn("DeleteObjects request failed; {} keys not deleted", + objectKeys.size() - deleted.size(), e); + break; + } + + for (final var deletedObject : response.deleted()) { + final ObjectKey key = byValue.get(deletedObject.key()); + if (key != null) { + deleted.add(key); } } - } catch (final ApiCallTimeoutException | ApiCallAttemptTimeoutException e) { - throw new StorageBackendTimeoutException("Failed to delete keys " + batch, e); - } catch (final SdkException e) { - throw new StorageBackendException("Failed to delete keys " + batch, e); + logDeleteErrors(response.errors()); } + return deleted; + } + + /** + * Logs per-key delete errors, distinguishing throttling (expected under load, aggregated) from + * hard errors (logged individually). No retry happens here: keys that were not deleted stay marked + * for deletion and are retried on the next FileCleaner cycle, while request-rate backoff is left to + * the S3 client's adaptive retry strategy. + */ + private void logDeleteErrors(final List errors) { + int throttled = 0; + for (final var error : errors) { + if (THROTTLE_ERROR_CODES.contains(error.code())) { + throttled++; + } else { + LOGGER.warn("Failed to delete {}: {} ({}); leaving it for the next cycle", + error.key(), error.message(), error.code()); + } + } + if (throttled > 0) { + LOGGER.info("{} keys throttled by S3; leaving them for the next cycle", throttled); + } + } + + private DeleteObjectsResponse deleteObjectsOnce(final Set keys) { + final Set ids = keys.stream() + .map(k -> ObjectIdentifier.builder().key(k.value()).build()) + .collect(Collectors.toSet()); + final Delete delete = Delete.builder().objects(ids).build(); + final DeleteObjectsRequest deleteObjectsRequest = DeleteObjectsRequest.builder() + .bucket(bucketName) + .delete(delete) + .build(); + return s3Client.deleteObjects(deleteObjectsRequest); } @Override diff --git a/storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java b/storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java index 42519ddef7a..648cffa231d 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/config/ConfigTestStorageBackend.java @@ -52,7 +52,8 @@ public void delete(ObjectKey key) throws StorageBackendException { } @Override - public void delete(Set keys) throws StorageBackendException { + public Set delete(Set keys) throws StorageBackendException { + return Set.copyOf(keys); } @Override diff --git a/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java b/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java index a9bbd67f52a..0e8a3b1f309 100644 --- a/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java +++ b/storage/inkless/src/test/java/io/aiven/inkless/delete/FileCleanerMockedTest.java @@ -40,7 +40,10 @@ import io.aiven.inkless.control_plane.DeleteFilesRequest; import io.aiven.inkless.control_plane.FileToDelete; import io.aiven.inkless.storage_backend.common.StorageBackend; +import io.aiven.inkless.storage_backend.common.StorageBackendException; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -75,11 +78,13 @@ void singleWithinRetention() throws Exception { final var now = TimeUtils.now(time); when(controlPlane.getFilesToDelete()) .thenReturn(List.of(new FileToDelete(objectKey.value(), now.minus(Duration.ofMinutes(15))))); + when(storageBackend.delete(Set.of(objectKey))).thenReturn(Set.of(objectKey)); cleaner.run(); verify(storageBackend, times(1)).delete(Set.of(objectKey)); verify(controlPlane, times(1)).deleteFiles(new DeleteFilesRequest(Set.of(objectKey.value()))); + assertEquals(0, cleaner.metrics.fileCleanerFilesFailed.sum()); } @Test @@ -105,10 +110,63 @@ void multiple() throws Exception { new FileToDelete(OBJECT_KEY_CREATOR.create("key2").value(), TimeUtils.now(time).minus(Duration.ofMinutes(5))), new FileToDelete(objectKeys.get(1).value(), TimeUtils.now(time).minus(Duration.ofMinutes(15))) )); + when(storageBackend.delete(new HashSet<>(objectKeys))).thenReturn(new HashSet<>(objectKeys)); cleaner.run(); verify(storageBackend, times(1)).delete(new HashSet<>(objectKeys)); verify(controlPlane, times(1)).deleteFiles(new DeleteFilesRequest(objectKeys.stream().map(ObjectKey::value).collect(Collectors.toSet()))); } + + @Test + void dereferencesOnlyKeysConfirmedDeleted() throws Exception { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD); + final var deleted = OBJECT_KEY_CREATOR.from("deleted"); + final var throttled = OBJECT_KEY_CREATOR.from("throttled"); + final var now = TimeUtils.now(time); + when(controlPlane.getFilesToDelete()) + .thenReturn(List.of( + new FileToDelete(deleted.value(), now.minus(Duration.ofMinutes(15))), + new FileToDelete(throttled.value(), now.minus(Duration.ofMinutes(15))) + )); + // Storage confirms only one key; the other was not deleted (e.g. throttled). + when(storageBackend.delete(Set.of(deleted, throttled))).thenReturn(Set.of(deleted)); + + cleaner.run(); + + // Only the confirmed key is dereferenced; the throttled one stays for the next cycle. + verify(controlPlane, times(1)).deleteFiles(new DeleteFilesRequest(Set.of(deleted.value()))); + assertEquals(1, cleaner.metrics.fileCleanerFilesFailed.sum()); + } + + @Test + void skipsControlPlaneWhenNothingDeleted() throws Exception { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD); + final var objectKey = OBJECT_KEY_CREATOR.from("key"); + final var now = TimeUtils.now(time); + when(controlPlane.getFilesToDelete()) + .thenReturn(List.of(new FileToDelete(objectKey.value(), now.minus(Duration.ofMinutes(15))))); + when(storageBackend.delete(Set.of(objectKey))).thenReturn(Set.of()); + + cleaner.run(); + + // Nothing drained (e.g. throttled): the control plane is left untouched so the keys stay marked + // for the next cycle. Cadence between cycles is set by the scheduler, not an in-run sleep. + verify(controlPlane, times(0)).deleteFiles(any()); + } + + @Test + void swallowsStorageFailureWithoutTouchingControlPlane() throws Exception { + final var cleaner = new FileCleaner(time, controlPlane, storageBackend, OBJECT_KEY_CREATOR, RETENTION_PERIOD); + final var objectKey = OBJECT_KEY_CREATOR.from("key"); + final var now = TimeUtils.now(time); + when(controlPlane.getFilesToDelete()) + .thenReturn(List.of(new FileToDelete(objectKey.value(), now.minus(Duration.ofMinutes(15))))); + when(storageBackend.delete(Set.of(objectKey))).thenThrow(new StorageBackendException("boom")); + + // run() catches the failure, records an error, and neither propagates nor touches the control plane. + cleaner.run(); + + verify(controlPlane, times(0)).deleteFiles(any()); + } } diff --git a/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/integration/S3StorageDeleteTest.java b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/integration/S3StorageDeleteTest.java new file mode 100644 index 00000000000..1834ed7e8ab --- /dev/null +++ b/storage/inkless/src/test/java/io/aiven/inkless/storage_backend/s3/integration/S3StorageDeleteTest.java @@ -0,0 +1,122 @@ +/* + * Inkless + * Copyright (C) 2024 - 2025 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 . + */ +package io.aiven.inkless.storage_backend.s3.integration; + +import org.apache.kafka.common.metrics.Metrics; + +import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo; +import com.github.tomakehurst.wiremock.junit5.WireMockTest; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +import io.aiven.inkless.common.ObjectKey; +import io.aiven.inkless.storage_backend.common.fixtures.TestObjectKey; +import io.aiven.inkless.storage_backend.s3.S3Storage; +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.regions.Region; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.anyUrl; +import static com.github.tomakehurst.wiremock.client.WireMock.exactly; +import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.postRequestedFor; +import static com.github.tomakehurst.wiremock.client.WireMock.stubFor; +import static com.github.tomakehurst.wiremock.client.WireMock.verify; +import static com.github.tomakehurst.wiremock.common.ContentTypes.CONTENT_TYPE; +import static org.apache.hc.core5.http.ContentType.APPLICATION_XML; +import static org.assertj.core.api.Assertions.assertThat; + +@Tag("integration") +@WireMockTest +class S3StorageDeleteTest { + private static final String BUCKET_NAME = "test-bucket"; + + private final Metrics metrics = new Metrics(); + private final S3Storage storage = new S3Storage(metrics); + + @AfterEach + void tearDown() throws Exception { + storage.close(); + } + + private void configure(final WireMockRuntimeInfo wmRuntimeInfo) { + storage.configure(Map.of( + "s3.bucket.name", BUCKET_NAME, + "s3.region", Region.US_EAST_1.id(), + "s3.endpoint.url", wmRuntimeInfo.getHttpBaseUrl(), + "s3.path.style.access.enabled", "true", + "aws.credentials.provider.class", AnonymousCredentialsProvider.class.getName() + )); + } + + private static String deleteResult(final String body) { + return "" + + "" + body + ""; + } + + private static String deleted(final String key) { + return "" + key + ""; + } + + private static String error(final String key, final String code) { + return "" + key + "" + code + "msg"; + } + + private static Set values(final Set keys) { + return keys.stream().map(ObjectKey::value).collect(Collectors.toSet()); + } + + @Test + void returnsOnlyConfirmedDeletionsAndDoesNotRetry(final WireMockRuntimeInfo wmRuntimeInfo) throws Exception { + configure(wmRuntimeInfo); + // key0 deleted; key1 throttled; key2 hard error. Neither failed key is retried in-call: + // both are left for the next FileCleaner cycle, and only the confirmed deletion is returned. + stubFor(post(anyUrl()) + .willReturn(aResponse().withStatus(200) + .withHeader(CONTENT_TYPE, APPLICATION_XML.getMimeType()) + .withBody(deleteResult( + deleted("key0") + error("key1", "SlowDown") + error("key2", "AccessDenied"))))); + + final Set deleted = storage.delete(Set.of( + new TestObjectKey("key0"), new TestObjectKey("key1"), new TestObjectKey("key2"))); + + assertThat(values(deleted)).containsExactlyInAnyOrder("key0"); + verify(exactly(1), postRequestedFor(anyUrl())); + } + + @Test + void doesNotThrowWhenWholeRequestFails(final WireMockRuntimeInfo wmRuntimeInfo) throws Exception { + configure(wmRuntimeInfo); + // A whole-request throttle (503) that the SDK's adaptive retry exhausts must not propagate: + // the keys are simply reported as not deleted and left for the next cycle. + stubFor(post(anyUrl()).willReturn(aResponse().withStatus(503) + .withHeader(CONTENT_TYPE, APPLICATION_XML.getMimeType()) + .withBody("SlowDownPlease reduce your request rate."))); + + final Set deleted = + storage.delete(Set.of(new TestObjectKey("key0"), new TestObjectKey("key1"))); + + assertThat(deleted).isEmpty(); + } +}