CompletableFuture<Void> clean(List<S3Object> objects) {
            List<S3Object> ignoredObjects = new LinkedList<>();
            List<S3Object> deepDeleteCompositeObjects = new LinkedList<>();
            List<S3Object> shallowDeleteObjects = new ArrayList<>(objects.size());
            for (S3Object object : objects) {
                ObjectAttributes attributes = ObjectAttributes.from(object.getAttributes());
                if (attributes.bucket() == LocalFileObjectStorage.BUCKET_ID) {
                    ignoredObjects.add(object);
                } else if (attributes.deepDelete() && attributes.type() == Type.Composite) {
                    deepDeleteCompositeObjects.add(object);
                } else {
                    shallowDeleteObjects.add(object);
                }
            }

            List<CompletableFuture<Void>> cfList = new LinkedList<>();
            // Delete the objects
            batchDelete(shallowDeleteObjects, this::shallowlyDelete, cfList);
            // Delete the composite object and it's linked objects
            batchDelete(deepDeleteCompositeObjects, this::deepDelete, cfList);
            // Delete the local file objects
            batchDelete(ignoredObjects, this::noopDelete, cfList);

            return CompletableFuture.allOf(cfList.toArray(new CompletableFuture[0]));
        }
Generate RemoveS3ObjectRecord for the deleted S3Objects.

@param deletedObjectIds the deleted S3Objects' ids
@return the result of the generation, contains the records which should be applied to the raft.
metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java