diff --git a/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java index 2f71a1c50a..c941a77020 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java @@ -69,6 +69,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.function.Supplier; @@ -89,6 +90,7 @@ public class S3ObjectControlManager { private static final long DEFAULT_LIFECYCLE_CHECK_INTERVAL_MS = 1000L; private static final long DEFAULT_INITIAL_DELAY_MS = 1000L; private static final int MAX_DELETE_BATCH_COUNT = 2000; + private static final int MAX_OBJECT_CLEAN_CONCURRENCY = 8; private final QuorumController quorumController; private final Logger log; @@ -541,6 +543,8 @@ private void removeMarkDestroyedObject(long objectId) { } class ObjectCleaner { + private final Semaphore cleanConcurrencyLimiter = new Semaphore(MAX_OBJECT_CLEAN_CONCURRENCY); + CompletableFuture clean(List objects) { List ignoredObjects = new LinkedList<>(); List deepDeleteCompositeObjects = new LinkedList<>(); @@ -558,26 +562,43 @@ CompletableFuture clean(List objects) { List> cfList = new LinkedList<>(); // Delete the objects - batchDelete(shallowDeleteObjects, this::shallowlyDelete, cfList); + throttledBatchDelete(shallowDeleteObjects, this::shallowlyDelete, cfList); // Delete the composite object and it's linked objects - batchDelete(deepDeleteCompositeObjects, this::deepDelete, cfList); + throttledBatchDelete(deepDeleteCompositeObjects, this::deepDelete, cfList); // Delete the local file objects - batchDelete(ignoredObjects, this::noopDelete, cfList); + throttledBatchDelete(ignoredObjects, this::noopDelete, cfList); return CompletableFuture.allOf(cfList.toArray(new CompletableFuture[0])); } - private void batchDelete(List objects, + /** + * Split objects into batches and delete each batch through the concurrency limiter. + * This prevents overwhelming S3 with too many concurrent delete requests when a + * large number of objects are destroyed at once (e.g. topic deletion or TTL change). + */ + private void throttledBatchDelete(List objects, Function, CompletableFuture> deleteFunc, List> cfList) { if (objects.isEmpty()) { return; } CollectionHelper.groupListByBatchSizeAsStream(objects, AwsObjectStorage.AWS_DEFAULT_BATCH_DELETE_OBJECTS_NUMBER) - .map(deleteFunc) + .map(batch -> throttled(batch, deleteFunc)) .forEach(cfList::add); } + private CompletableFuture throttled(List batch, + Function, CompletableFuture> deleteFunc) { + try { + cleanConcurrencyLimiter.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return CompletableFuture.failedFuture(new RuntimeException(e)); + } + return deleteFunc.apply(batch) + .whenComplete((v, ex) -> cleanConcurrencyLimiter.release()); + } + private CompletableFuture shallowlyDelete(List s3objects) { List objectPaths = s3objects.stream().map(o -> new ObjectPath(o.bucket(), o.getObjectKey())).collect(Collectors.toList()); return objectStorage.delete(objectPaths) @@ -591,14 +612,28 @@ private CompletableFuture shallowlyDelete(List s3objects) { }); } + /** + * Deep-delete composite objects with bounded concurrency. Each composite + * delete issues multiple S3 API calls (read index, delete linked objects, + * delete self), so individual objects are throttled through the shared + * concurrency limiter to prevent S3 rate-limiting. + */ private CompletableFuture deepDelete(List s3Objects) { if (s3Objects.isEmpty()) { return CompletableFuture.completedFuture(null); } - List> cfList = new LinkedList<>(); + List> cfList = new ArrayList<>(s3Objects.size()); for (S3Object object : s3Objects) { S3ObjectMetadata metadata = new S3ObjectMetadata(object.getObjectId(), object.getAttributes()); - cfList.add(CompositeObject.delete(metadata, objectStorage)); + try { + cleanConcurrencyLimiter.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + cfList.add(CompletableFuture.failedFuture(new RuntimeException(e))); + break; + } + CompletableFuture deleteCf = CompositeObject.delete(metadata, objectStorage); + cfList.add(deleteCf.whenComplete((v, ex) -> cleanConcurrencyLimiter.release())); } CompletableFuture allCf = CompletableFuture.allOf(cfList.toArray(new CompletableFuture[0])); allCf.thenAccept(rst -> { diff --git a/metadata/src/test/java/org/apache/kafka/controller/stream/S3ObjectControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/stream/S3ObjectControlManagerTest.java index 0351a78c17..e76ed95e3c 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/stream/S3ObjectControlManagerTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/stream/S3ObjectControlManagerTest.java @@ -376,4 +376,63 @@ private long prepareOneObject(long ttl) { return objectId; } + /** + * Given 20 committed-then-mark-destroyed objects, when the ObjectCleaner runs, + * verify that S3 delete requests never exceed MAX_OBJECT_CLEAN_CONCURRENCY (8) + * concurrent calls. + */ + @Test + public void testCleanConcurrencyLimit() throws Exception { + Mockito.when(controller.checkS3ObjectsLifecycle(any(ControllerRequestContext.class))) + .then(inv -> { + ControllerResult result = manager.checkS3ObjectsLifecycle(); + replay(manager, result.records()); + return CompletableFuture.completedFuture(null); + }); + Mockito.when(controller.notifyS3ObjectDeleted(any(ControllerRequestContext.class), anyList())) + .then(inv -> { + ControllerResult result = manager.notifyS3ObjectDeleted(inv.getArgument(1)); + replay(manager, result.records()); + return CompletableFuture.completedFuture(null); + }); + + // Track concurrent delete calls + java.util.concurrent.atomic.AtomicInteger concurrentDeletes = new java.util.concurrent.atomic.AtomicInteger(0); + java.util.concurrent.atomic.AtomicInteger maxConcurrentDeletes = new java.util.concurrent.atomic.AtomicInteger(0); + List> deleteFutures = java.util.Collections.synchronizedList(new java.util.ArrayList<>()); + + Mockito.when(objectStorage.delete(anyList())).then(inv -> { + int current = concurrentDeletes.incrementAndGet(); + maxConcurrentDeletes.updateAndGet(max -> Math.max(max, current)); + CompletableFuture cf = new CompletableFuture<>(); + deleteFutures.add(cf); + cf.whenComplete((v, ex) -> concurrentDeletes.decrementAndGet()); + return cf; + }); + + // Create 20 mark-destroyed objects + int objectCount = 20; + for (int i = 0; i < objectCount; i++) { + long objectId = prepareOneObject(60 * 1000); + ControllerResult commitResult = manager.commitObject(objectId, 1024, 1313L, ObjectAttributes.DEFAULT.attributes()); + replay(manager, commitResult.records()); + ControllerResult destroyResult = manager.markDestroyObjects(List.of(objectId)); + replay(manager, destroyResult.records()); + } + + // Trigger lifecycle check to start deletion + time.sleep(1100L); + manager.checkS3ObjectsLifecycle(); + time.sleep(1100L); + manager.checkS3ObjectsLifecycle(); + + // The max concurrent deletes should not exceed 8 (MAX_OBJECT_CLEAN_CONCURRENCY) + assertTrue(maxConcurrentDeletes.get() <= 8, + "Max concurrent deletes " + maxConcurrentDeletes.get() + " should not exceed 8"); + assertTrue(maxConcurrentDeletes.get() > 0, "Should have had at least one delete call"); + + // Complete all pending deletes + deleteFutures.forEach(cf -> cf.complete(null)); + } + }