public ControllerResult<Void> checkS3ObjectsLifecycle() {
        if (!lastCleanCf.isDone()) {
            log.info("the last deletion[timestamp={}] is not finished, skip this check", lastCleanStartTimestamp);
            return ControllerResult.of(Collections.emptyList(), null);
        }
        AutoMQVersion version = this.version.get();
        short objectRecordVersion = version.objectRecordVersion();
        List<ApiMessageAndVersion> records = new ArrayList<>();
        List<Long> ttlReachedObjects = new LinkedList<>();
        long now = time.milliseconds();

        // check the prepared object deadline
        for (int i = 0; i < MAX_DELETE_BATCH_COUNT; i++) {
            // only allow checking MAX_DELETE_BATCH_COUNT objects in one round to avoid blocking the event loop
            Long waitingDeadlineCheckPreparedObject = waitingDeadlineCheckPreparedObjects.poll();
            if (waitingDeadlineCheckPreparedObject == null) {
                break;
            }
            preparedObjectsTimeouts.remove(waitingDeadlineCheckPreparedObject);
            S3Object obj = objectsMetadata.get(waitingDeadlineCheckPreparedObject);
            if (obj == null) {
                continue;
            }
            if (obj.isExpired(now)) {
                S3ObjectRecord record = new S3ObjectRecord()
                    .setObjectId(obj.getObjectId())
                    .setObjectState((byte) S3ObjectState.MARK_DESTROYED.ordinal())
                    .setObjectSize(obj.getObjectSize());
                if (version.isHugeClusterSupported()) {
                    record.setTimestamp(0);
                } else {
                    record.setMarkDestroyedTimeInMs(obj.getTimestamp());
                }
                if (version.isObjectAttributesSupported()) {
                    record.setAttributes(ObjectAttributes.builder().bucket(ObjectAttributes.MATCH_ALL_BUCKET).build().attributes());
                }
                ttlReachedObjects.add(obj.getObjectId());
                // generate the records which mark the expired objects as destroyed
                records.add(new ApiMessageAndVersion(record, objectRecordVersion));
            } else {
                // reschedule the deadline check
                preparedObjectsTimeouts.put(obj.getObjectId(), newPreparedObjectTimeout(obj.getObjectId(), obj.getTimestamp()));
            }
        }
        if (now - lastPreparedObjectsSwitchTimestamp > TimeUnit.MINUTES.toMillis(60)) {
            // The preparedObjectsTimeouts isn't a Timeline struct, so it's not consistent with the preparedObjects0/1.
            // Consider the following scenario:
            // 1. Prepare(obj1), add timeout check.
            // 2. Commit(obj1), cancel timeout check.
            // 3. The image is reset back to step 1, and no commit.
            // 4. The obj1 timeout check is missing.
            // So we try to add missing timeout check when swap the preparedObjects0/1.
            TimelineHashMap<Long, Long> tmp = preparedObjects1;
            preparedObjects1 = preparedObjects0;
            preparedObjects0 = tmp;
            tmp.forEach((objectId, deadline) ->
                preparedObjectsTimeouts.computeIfAbsent(objectId, id -> newPreparedObjectTimeout(objectId, deadline))
            );
            lastPreparedObjectsSwitchTimestamp = now;
        }
        if (!ttlReachedObjects.isEmpty()) {
            log.info("objects TTL is reached, objects={}", ttlReachedObjects);
        }

        // The mark destroyed object will be truly deleted after the [retention time, 2 * retention time]
        if (now - lastMarkDestroyedObjectsSwitchTimestamp > this.config.objectRetentionTimeInSecond() * 1000L) {
            TimelineHashSet<Long> expired = this.markDestroyedObjects1;
            boolean swap = true;
            if (!expired.isEmpty()) {
                List<Long> fastDeleteObjects = new ArrayList<>();
                int expiredSize = expired.size();
                if (expiredSize > MAX_DELETE_BATCH_COUNT) {
                    swap = false;
                }
                List<S3Object> requiredDeleteObjects = new ArrayList<>(Math.min(expiredSize, MAX_DELETE_BATCH_COUNT));
                int deleteCount = 0;
                for (Long objectId : expired) {
                    S3Object s3Object = objectsMetadata.get(objectId);
                    if (s3Object == null) {
                        fastDeleteObjects.add(objectId);
                    } else {
                        requiredDeleteObjects.add(s3Object);
                    }
                    if (++deleteCount >= MAX_DELETE_BATCH_COUNT) {
                        // only allow checking MAX_DELETE_BATCH_COUNT objects in one round to avoid blocking the event loop
                        break;
                    }
                }
                fastDeleteObjects.forEach(expired::remove);
                this.lastCleanCf = this.objectCleaner.clean(requiredDeleteObjects);
                this.lastCleanStartTimestamp = now;
            }
            if (swap) {
                // When the markDestroyedObjects1 could be completely deleted in MAX_DELETE_BATCH_COUNT, then swap the markDestroyedObjects0/1.
                this.markDestroyedObjects1 = markDestroyedObjects0;
                this.markDestroyedObjects0 = expired;
                this.lastMarkDestroyedObjectsSwitchTimestamp = now;
            }
        }
        return ControllerResult.of(records, null);
    }
Check the S3Object's lifecycle, mark destroy the expired objects and trigger truly deletion for the mark destroyed objects.

@return the result of the check, contains the records which should be applied to the raft.
metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java