8000 fix(inkless:storage): bound bulk-delete failure logging [KC-430] by jeqo · Pull Request #742 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

fix(inkless:storage): bound bulk-delete failure logging [KC-430] #742

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jeqo merged 1 commit into from
Aug 19, 2026
Merged

fix(inkless:storage): bound bulk-delete failure logging [KC-430] #742

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 27 additions & 2 deletions storage/inkless/src/main/java/io/aiven/inkless/storage_backend/azure/AzureBlobStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import com.azure.identity.DefaultAzureCredentialBuilder;
import com.azure.storage.blob.BlobContainerClient;
import com.azure.storage.blob.BlobServiceClientBuilder;
import com.azure.storage.blob.models.BlobErrorCode;
import com.azure.storage.blob.models.BlobRange;
import com.azure.storage.blob.models.BlobStorageException;
import com.azure.storage.blob.models.ParallelTransferOptions;
Expand All @@ -45,6 +46,7 @@
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;

import io.aiven.inkless.common.ByteRange;
import io.aiven.inkless.common.ObjectKey;
Expand All @@ -58,6 +60,11 @@
public final class AzureBlobStorage extends StorageBackend {
private static final Logger LOGGER = LoggerFactory.getLogger(AzureBlobStorage.class);

// Azure error codes that indicate throttling rather than a hard failure. The set only picks the log
// level: the next FileCleaner cycle retries either kind. INTERNAL_ERROR and OPERATION_TIMED_OUT stay
// out because a transient server error doesn't self-heal like backpressure, so they keep the WARN.
private static final Set<String> THROTTLE_ERROR_CODES = Set.of(BlobErrorCode.SERVER_BUSY.toString());

private AzureBlobStorageConfig config;
private BlobContainerClient blobContainerClient;
private MetricCollector.MetricsPolicy policy;
Expand Down Expand Up @@ -207,16 +214,34 @@ public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendExc
// 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<>();
// Count the failures by error code instead of logging one line per key: a pass that fails for
// every key repeats on every FileCleaner cycle, so per 8000 -key lines grow with the worklist.
final Map<String, Integer> failuresByCode = new TreeMap<>();
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);
failuresByCode.merge(String.valueOf(e.getErrorCode()), 1, Integer::sum);
} catch (final RuntimeException e) {
LOGGER.warn("Failed to delete {}; leaving it for the next cycle", key, Exceptions.unwrap(e));
// Not a service response, so there is no error code to count, and nothing says the next key
// fares better: the client itself is likely unusable. Report it in full and stop the pass,
// since deletion is idempotent and the remaining keys retry on the next FileCleaner cycle.
LOGGER.warn("Deleting {} failed unexpectedly, stopping the pass", key, Exceptions.unwrap(e));
break;
}
}

if (!failuresByCode.isEmpty()) {
// Throttling is backpressure the next FileCleaner cycle retries; any other code needs an
// operator, so it must not sit at INFO while the worklist stops draining.
if (THROTTLE_ERROR_CODES.containsAll(failuresByCode.keySet())) {
LOGGER.info("Delete failures by error code {}", failuresByCode);
} else {
LOGGER.warn("Delete failures by error code {}", failuresByCode);
}
}

return deleted;
}

Expand Down
2 changes: 1 addition & 1 deletion storage/inkless/src/main/java/io/aiven/inkless/storage_backend/gcs/GcsStorage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendExc
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);
}
}

Expand Down
55 changes: 26 additions & 29 deletions storage/inkless/src/main/java/io/aiven/inkless/storage_backend/s3/S3Storage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;

import io.aiven.inkless.common.ByteRange;
Expand All @@ -58,7 +59,6 @@
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 {
Expand All @@ -67,10 +67,13 @@ public final class S3Storage extends StorageBackend {

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<String> THROTTLE_ERROR_CODES =
Set.of("SlowDown", "ServiceUnavailable", "RequestLimitExceeded");
// Per-key S3 error codes that indicate throttling rather than a hard failure. Matched by name because
// a per-key error is not an exception, so the SDK's own check (RetryUtils.isThrottlingException) does
// not apply; both names appear in the SDK's throttling list, AwsErrorCode.THROTTLING_ERROR_CODES,
// which is internal API and so not referenced here. ServiceUnavailable stays out: AWS classifies it as
// a retryable 503 rather than throttling, and a transient server error doesn't self-heal the way
// backpressure does, so it keeps the WARN.
private static final Set<String> THROTTLE_ERROR_CODES = Set.of("SlowDown", "RequestLimitExceeded");

private S3Client s3Client;
private String bucketName;
Expand Down Expand Up @@ -171,6 +174,10 @@ public void delete(final ObjectKey key) throws StorageBackendException {
public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendException {
final List<ObjectKey> objectKeys = new ArrayList<>(keys);
final Set<ObjectKey> deleted = new HashSet<>();
// Count the failures by error code instead of logging one line per key: a pass that fails for
// every key repeats on every FileCleaner cycle, so per-key lines grow with the worklist.
final Map<String, Integer> failuresByCode = new TreeMap<>();

for (int i = 0; i < objectKeys.size(); i += MAX_DELETE_KEYS_LIMIT) {
final Set<ObjectKey> batch = new HashSet<>(objectKeys.subList(
i,
Expand All @@ -182,11 +189,9 @@ public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendExc
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);
// Nothing in this batch was attempted. Stop the pass; deletion is idempotent, so
// retrying the keys later is safe.
LOGGER.warn("DeleteObjects request failed, stopping the pass", e);
break;
}

Expand All @@ -196,30 +201,22 @@ public Set<ObjectKey> delete(final Set<ObjectKey> keys) throws StorageBackendExc
deleted.add(key);
}
}
logDeleteErrors(response.errors());
for (final var error : response.errors()) {
failuresByCode.merge(error.code(), 1, Integer::sum);
}
}
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<S3Error> errors) {
int throttled = 0;
for (final var error : errors) {
if (THROTTLE_ERROR_CODES.contains(error.code())) {
throttled++;
if (!failuresByCode.isEmpty()) {
// Throttling is backpressure the next FileCleaner cycle retries; any other code needs an
// operator, so it must not sit at INFO while the worklist stops draining.
if (THROTTLE_ERROR_CODES.containsAll(failuresByCode.keySet())) {
LOGGER.info("Delete failures by error code {}", failuresByCode);
} else {
LOGGER.warn("Failed to delete {}: {} ({}); leaving it for the next cycle",
error.key(), error.message(), error.code());
LOGGER.warn("Delete failures by error code {}", failuresByCode);
}
}
if (throttled > 0) {
LOGGER.info("{} keys throttled by S3; leaving them for the next cycle", throttled);
}

return deleted;
}

private DeleteObjectsResponse deleteObjectsOnce(final Set<ObjectKey> keys) {
Expand Down
Loading
0