feat(s3): throttle ObjectCleaner deletion concurrency to prevent S3 rate-limiting (#3203)

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<Void> clean(List<S3Object> objects) {
             List<S3Object> ignoredObjects = new LinkedList<>();
             List<S3Object> deepDeleteCompositeObjects = new LinkedList<>();
@@ -558,26 +562,43 @@ CompletableFuture<Void> clean(List<S3Object> objects) {
 
             List<CompletableFuture<Void>> 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<S3Object> 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<S3Object> objects,
             Function<List<S3Object>, CompletableFuture<Void>> deleteFunc,
             List<CompletableFuture<Void>> 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<Void> throttled(List<S3Object> batch,
+            Function<List<S3Object>, CompletableFuture<Void>> 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<Void> shallowlyDelete(List<S3Object> s3objects) {
             List<ObjectPath> objectPaths = s3objects.stream().map(o -> new ObjectPath(o.bucket(), o.getObjectKey())).collect(Collectors.toList());
             return objectStorage.delete(objectPaths)
@@ -591,14 +612,28 @@ private CompletableFuture<Void> shallowlyDelete(List<S3Object> 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<Void> deepDelete(List<S3Object> s3Objects) {
             if (s3Objects.isEmpty()) {
                 return CompletableFuture.completedFuture(null);
             }
-            List<CompletableFuture<Void>> cfList = new LinkedList<>();
+            List<CompletableFuture<Void>> 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<Void> deleteCf = CompositeObject.delete(metadata, objectStorage);
+                cfList.add(deleteCf.whenComplete((v, ex) -> cleanConcurrencyLimiter.release()));
             }
             CompletableFuture<Void> allCf = CompletableFuture.allOf(cfList.toArray(new CompletableFuture[0]));
             allCf.thenAccept(rst -> {

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<Void> result = manager.checkS3ObjectsLifecycle();
+                replay(manager, result.records());
+                return CompletableFuture.completedFuture(null);
+            });
+        Mockito.when(controller.notifyS3ObjectDeleted(any(ControllerRequestContext.class), anyList()))
+            .then(inv -> {
+                ControllerResult<Void> 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<CompletableFuture<Void>> 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<Void> 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<Errors> commitResult = manager.commitObject(objectId, 1024, 1313L, ObjectAttributes.DEFAULT.attributes());
+            replay(manager, commitResult.records());
+            ControllerResult<Boolean> 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));
+    }
+
 }