public ControllerResult<Boolean> markDestroyObjects(List<Long> objects, List<CompactOperations> operations) {
        AutoMQVersion version = this.version.get();
        short objectRecordVersion = version.objectRecordVersion();
        List<ApiMessageAndVersion> records = new ArrayList<>();
        long now = time.milliseconds();
        for (int i = 0; i < objects.size(); i++) {
            Long objectId = objects.get(i);
            CompactOperations operation = operations.get(i);
            S3Object object = this.objectsMetadata.get(objectId);
            if (object == null || object.getS3ObjectState() == S3ObjectState.MARK_DESTROYED) {
                log.error("object {} not exist when mark destroy object", objectId);
                return ControllerResult.of(Collections.emptyList(), false);
            }
            int attributes = object.getAttributes();
            if (object.getS3ObjectState() == S3ObjectState.PREPARED) {
                attributes = ObjectAttributes.builder(attributes).bucket(ObjectAttributes.MATCH_ALL_BUCKET).build().attributes();
            }

            switch (operation) {
                case DELETE: {
                    S3ObjectRecord record = new S3ObjectRecord()
                        .setObjectId(objectId)
                        .setObjectSize(object.getObjectSize())
                        .setObjectState(S3ObjectState.MARK_DESTROYED.toByte());
                    if (version.isHugeClusterSupported()) {
                        record.setTimestamp(now);
                    } else {
                        record.setMarkDestroyedTimeInMs(now);
                    }
                    if (version.isCompositeObjectSupported()) {
                        record.setAttributes(attributes);
                    }
                    records.add(new ApiMessageAndVersion(record, objectRecordVersion));
                    break;
                }
                case KEEP_DATA: {
                    records.add(new ApiMessageAndVersion(new RemoveS3ObjectRecord().setObjectId(objectId), (short) 0));
                    break;
                }
                case DEEP_DELETE: {
                    S3ObjectRecord record = new S3ObjectRecord()
                        .setObjectId(objectId)
                        .setObjectSize(object.getObjectSize())
                        .setObjectState(S3ObjectState.MARK_DESTROYED.toByte());
                    if (version.isHugeClusterSupported()) {
                        record.setTimestamp(now);
                    } else {
                        record.setMarkDestroyedTimeInMs(now);
                    }
                    if (version.isCompositeObjectSupported()) {
                        int newAttributes = ObjectAttributes.builder(attributes).deepDelete().build().attributes();
                        record.setAttributes(newAttributes);
                    }
                    records.add(new ApiMessageAndVersion(record, objectRecordVersion));
                    break;
                }
                default:
                    throw new IllegalArgumentException("Unknown operation: " + operation);
            }
        }
        return ControllerResult.atomicOf(records, true);
    }
The objectId of the next object to be prepared. (start from 0)
metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java