feat(issues1171): improve objects deletion performance (#1173)

Signed-off-by: Robin Han <[email protected]>
AutoMQ · Apr 24, 2024 · 31f623d · 31f623d
1 parent de08a8b
commit 31f623d
Showing 1 changed file with 12 additions and 9 deletions.
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
@@ -257,7 +257,6 @@ public void replay(RemoveS3ObjectRecord record) {
         if (object != null) {
             s3ObjectSize.addAndGet(-object.getObjectSize());
         }
-        markDestroyedObjects.remove(record.objectId());
         preparedObjects.remove(record.objectId());
     }
 
@@ -300,23 +299,26 @@ public ControllerResult<Void> checkS3ObjectsLifecycle() {
         }
         // check the mark destroyed objects
         List<String> requiredDeleteKeys = new LinkedList<>();
-        List<Long> notExistObjects = new LinkedList<>();
-        for (Long objectId : this.markDestroyedObjects) {
+        while (true) {
+            Long objectId = this.markDestroyedObjects.peek();
+            if (objectId == null) {
+                break;
+            }
             S3Object object = this.objectsMetadata.get(objectId);
             if (object == null) {
-                notExistObjects.add(objectId);
+                // markDestroyedObjects isn't Timeline structure, so it may contains dirty / duplicated objectId
+                this.markDestroyedObjects.poll();
                 continue;
             }
             if (object.getMarkDestroyedTimeInMs() + (this.config.objectRetentionTimeInSecond() * 1000L) < System.currentTimeMillis()) {
+                this.markDestroyedObjects.poll();
                 // exceed delete retention time, trigger the truly deletion
                 requiredDeleteKeys.add(object.getObjectKey());
             } else {
                 // the following objects' mark destroyed time is not expired, so break the loop
                 break;
             }
         }
-        // markDestroyedObjects isn't Timeline structure, so it may contains dirty / duplicated objectId
-        notExistObjects.forEach(this.markDestroyedObjects::remove);
 
         if (!requiredDeleteKeys.isEmpty()) {
             this.lastCleanStartTimestamp = System.currentTimeMillis();
@@ -332,9 +334,10 @@ public ControllerResult<Void> checkS3ObjectsLifecycle() {
      * @return the result of the generation, contains the records which should be applied to the raft.
      */
     public ControllerResult<Void> notifyS3ObjectDeleted(List<Long> deletedObjectIds) {
-        List<ApiMessageAndVersion> records = deletedObjectIds.stream().filter(markDestroyedObjects::contains)
-                .map(objectId -> new ApiMessageAndVersion(new RemoveS3ObjectRecord()
-                        .setObjectId(objectId), (short) 0)).collect(Collectors.toList());
+        List<ApiMessageAndVersion> records = deletedObjectIds
+            .stream()
+            .map(objectId -> new ApiMessageAndVersion(new RemoveS3ObjectRecord().setObjectId(objectId), (short) 0))
+            .collect(Collectors.toList());
         return ControllerResult.of(records, null);
     }