public ControllerResult<PrepareS3ObjectResponseData> prepareObject(PrepareS3ObjectRequestData request) {
        AutoMQVersion version = this.version.get();
        short objectRecordVersion = version.objectRecordVersion();
        // TODO: pre assigned a batch of objectIds in controller
        List<ApiMessageAndVersion> records = new ArrayList<>();
        PrepareS3ObjectResponseData response = new PrepareS3ObjectResponseData();
        int count = request.preparedCount();

        // update assigned stream id
        long newAssignedObjectId = nextAssignedObjectId.get() + count - 1;
        records.add(new ApiMessageAndVersion(new AssignedS3ObjectIdRecord()
            .setAssignedS3ObjectId(newAssignedObjectId), (short) 0));

        long firstAssignedObjectId = nextAssignedObjectId.get();
        long now = time.milliseconds();
        for (int i = 0; i < count; i++) {
            long objectId = nextAssignedObjectId.get() + i;
            long expiredTs = now + request.timeToLiveInMs();
            S3ObjectRecord record = new S3ObjectRecord()
                .setObjectId(objectId)
                .setObjectState(S3ObjectState.PREPARED.toByte());
            if (version.isHugeClusterSupported()) {
                record.setTimestamp(expiredTs);
            } else {
                record.setPreparedTimeInMs(now)
                    .setExpiredTimeInMs(expiredTs);
            }
            records.add(new ApiMessageAndVersion(record, objectRecordVersion));
        }
        response.setFirstS3ObjectId(firstAssignedObjectId);
        return ControllerResult.atomicOf(records, response);
    }
The objectId of the next object to be prepared. (start from 0)
metadata/src/main/java/org/apache/kafka/controller/stream/S3ObjectControlManager.java