public ControllerResult<Errors> commitObject(long objectId, long objectSize, long committedTs, int attributes) {
        if (objectId == NOOP_OBJECT_ID) {
            return ControllerResult.of(Collections.emptyList(), Errors.NONE);
        }
        S3Object object = this.objectsMetadata.get(objectId);
        if (object == null) {
            log.error("object {} not exist when commit object", objectId);
            return ControllerResult.of(Collections.emptyList(), Errors.OBJECT_NOT_EXIST);
        }
        // verify the state
        if (object.getS3ObjectState() == S3ObjectState.COMMITTED) {
            log.warn("object {} already committed", objectId);
            return ControllerResult.of(Collections.emptyList(), Errors.REDUNDANT_OPERATION);
        }
        if (object.getS3ObjectState() != S3ObjectState.PREPARED) {
            log.error("object {} is not prepared but try to commit", objectId);
            return ControllerResult.of(Collections.emptyList(), Errors.OBJECT_NOT_EXIST);
        }
        AutoMQVersion version = this.version.get();
        S3ObjectRecord record = new S3ObjectRecord()
            .setObjectId(objectId)
            .setObjectSize(objectSize)
            .setObjectState(S3ObjectState.COMMITTED.toByte());
        if (version.isHugeClusterSupported()) {
            record.setTimestamp(committedTs);
        } else {
            record.setCommittedTimeInMs(object.getTimestamp());
        }
        if (version.isObjectAttributesSupported()) {
            record.setAttributes(attributes);
        }
        return ControllerResult.of(List.of(
            new ApiMessageAndVersion(record, version.objectRecordVersion())), Errors.NONE);
    }
The objectId of the next object to be prepared. (start from 0)
metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java