public ControllerResult<CreateStreamResponse> createStream(int nodeId, long nodeEpoch,
        CreateStreamRequest request) {
        CreateStreamResponse resp = new CreateStreamResponse();
        // verify node epoch
        Errors nodeEpochCheckResult = nodeEpochCheck(nodeId, nodeEpoch);
        if (nodeEpochCheckResult != Errors.NONE) {
            resp.setErrorCode(nodeEpochCheckResult.code());
            log.warn("[CreateStream] invalid node epoch. nodeId={}, nodeEpoch={}, error={}",
                nodeId, nodeEpoch, nodeEpochCheckResult);
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // TODO: pre assigned a batch of stream id in controller
        long streamId = nextAssignedStreamId.get();
        // update assigned id
        ApiMessageAndVersion record0 = new ApiMessageAndVersion(new AssignedStreamIdRecord()
            .setAssignedStreamId(streamId), (short) 0);
        // create stream
        S3StreamRecord s3StreamRecord = new S3StreamRecord()
            .setStreamId(streamId)
            .setEpoch(S3StreamConstant.INIT_EPOCH)
            .setStartOffset(S3StreamConstant.INIT_START_OFFSET)
            .setRangeIndex(S3StreamConstant.INIT_RANGE_INDEX);
        AutoMQVersion autoMQVersion = featureControlManager.autoMQVersion();
        if (autoMQVersion.isStreamTagsSupported()) {
            S3StreamRecord.TagCollection tags = new S3StreamRecord.TagCollection();
            request.tags().forEach(tag -> tags.add(new S3StreamRecord.Tag().setKey(tag.key()).setValue(tag.value())));
            s3StreamRecord.setTags(tags);
        }
        ApiMessageAndVersion record = new ApiMessageAndVersion(s3StreamRecord, autoMQVersion.streamRecordVersion());
        resp.setStreamId(streamId);
        log.info("[CreateStream] successfully create a stream. streamId={}, nodeId={}, nodeEpoch={}", streamId, nodeId, nodeEpoch);
        return ControllerResult.atomicOf(Arrays.asList(record0, record), resp);
    }
The next stream id to be assigned.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java