public ControllerResult<OpenStreamResponse> openStream(int nodeId, long nodeEpoch, OpenStreamRequest request) {
        OpenStreamResponse resp = new OpenStreamResponse();
        long streamId = request.streamId();
        long epoch = request.streamEpoch();

        AutoMQVersion version = featureControlManager.autoMQVersion();

        // verify node epoch
        Errors nodeEpochCheckResult = nodeEpochCheck(nodeId, nodeEpoch);
        if (nodeEpochCheckResult != Errors.NONE) {
            resp.setErrorCode(nodeEpochCheckResult.code());
            log.warn("[OpenStream] invalid node epoch. streamId={}, nodeId={}, nodeEpoch={}, error={}",
                streamId, nodeId, nodeEpoch, nodeEpochCheckResult);
            return ControllerResult.of(Collections.emptyList(), resp);
        }

        // verify stream exist
        if (!this.streamsMetadata.containsKey(streamId)) {
            resp.setErrorCode(Errors.STREAM_NOT_EXIST.code());
            log.warn("[OpenStream] stream not exist. streamId={}, nodeId={}, nodeEpoch={}", streamId, nodeId, nodeEpoch);
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // verify epoch match
        StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamId);
        if (streamMetadata.currentEpoch() > epoch) {
            resp.setErrorCode(Errors.STREAM_FENCED.code());
            log.warn("[OpenStream] steam has been fenced. streamId={}, streamEpoch={}, requestEpoch={}, nodeId={}, nodeEpoch={}",
                streamId, streamMetadata.currentEpoch(), epoch, nodeId, nodeEpoch);
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        if (streamMetadata.currentEpoch() == epoch) {
            return retryOpen(streamMetadata, nodeId, nodeEpoch, request, resp, version);
        }
        if (streamMetadata.currentState() == StreamState.OPENED) {
            // the stream still in opened state, so it can't open until it is closed
            log.warn("[OpenStream] stream still in opened state. streamId={}, streamEpoch={}, ownerId={}, nodeId={}, nodeEpoch={}",
                streamId, streamMetadata.currentEpoch(), streamMetadata.currentRangeOwner(), nodeId, nodeEpoch);
            resp.setErrorCode(Errors.STREAM_NOT_CLOSED.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        int currentRangeOwner = streamMetadata.currentRangeOwner();
        if (nodeId != currentRangeOwner && lockedNodes.contains(currentRangeOwner)) {
            // Forbidden other nodes to open the stream if the last range is owned by a locked node
            resp.setErrorCode(Errors.NODE_LOCKED.code());
            log.warn("[OpenStream] the stream's last range is owned by a locked node {}. streamId={}, streamEpoch={}, requestEpoch={}, nodeId={}, nodeEpoch={}",
                currentRangeOwner, streamId, streamMetadata.currentEpoch(), epoch, nodeId, nodeEpoch);
            tryReassignPartitionBack(streamMetadata);
            return ControllerResult.of(Collections.emptyList(), resp);
        }

        // now the request is valid, update the stream's epoch and create a new range for this node
        List<ApiMessageAndVersion> records = new ArrayList<>();
        int newRangeIndex = streamMetadata.currentRangeIndex() + 1;
        // stream update record
        AutoMQVersion autoMQVersion = featureControlManager.autoMQVersion();
        S3StreamRecord s3StreamRecord = new S3StreamRecord()
            .setStreamId(streamId)
            .setEpoch(epoch)
            .setRangeIndex(newRangeIndex)
            .setStartOffset(streamMetadata.startOffset())
            .setStreamState(StreamState.OPENED.toByte());
        if (request.tags().size() > 0 && autoMQVersion.isStreamTagsSupported()) {
            // Compatible with the stream created in the old version, add missing tags for the stream.
            S3StreamRecord.TagCollection tags = new S3StreamRecord.TagCollection();
            request.tags().forEach(tag -> tags.add(new S3StreamRecord.Tag().setKey(tag.key()).setValue(tag.value())));
            s3StreamRecord.setTags(tags);
        }
        records.add(new ApiMessageAndVersion(s3StreamRecord, autoMQVersion.streamRecordVersion()));
        // get new range's start offset
        // default regard this range is the first range in stream, use 0 as start offset
        long nextRangeStartOffset = 0;
        if (newRangeIndex > 0) {
            // means that the new range is not the first range in stream, get the last range's end offset
            RangeMetadata lastRangeMetadata = streamMetadata.ranges().get(streamMetadata.currentRangeIndex());
            // the RangeMetadata in S3StreamMetadataImage is only update when create, rollToNext and trim
            records.add(new ApiMessageAndVersion(new RangeRecord()
                .setStreamId(streamId)
                .setNodeId(lastRangeMetadata.nodeId())
                .setStartOffset(lastRangeMetadata.startOffset())
                .setEndOffset(streamMetadata.endOffset())
                .setEpoch(lastRangeMetadata.epoch())
                .setRangeIndex(lastRangeMetadata.rangeIndex()), (short) 0));
            nextRangeStartOffset = streamMetadata.endOffset();
        }
        // Fix https://github.com/AutoMQ/automq/issues/1222
        // Handle the case: trim offset beyond the range end offset
        nextRangeStartOffset = Math.max(streamMetadata.startOffset(), nextRangeStartOffset);
        // range create record
        records.add(new ApiMessageAndVersion(new RangeRecord()
            .setStreamId(streamId)
            .setNodeId(nodeId)
            .setStartOffset(nextRangeStartOffset)
            .setEndOffset(nextRangeStartOffset)
            .setEpoch(epoch)
            .setRangeIndex(newRangeIndex), (short) 0));
        resp.setStartOffset(streamMetadata.startOffset());
        resp.setNextOffset(nextRangeStartOffset);

        for (RangeMetadata removableRange : streamMetadata.checkRemovableRanges()) {
            records.add(new ApiMessageAndVersion(new RemoveRangeRecord()
                .setStreamId(streamId)
                .setRangeIndex(removableRange.rangeIndex()), (short) 0));
        }

        log.info("[OpenStream] successfully open the stream. streamId={}, streamEpoch={}, nodeId={}, nodeEpoch={}",
            streamId, epoch, nodeId, nodeEpoch);
        return ControllerResult.atomicOf(records, resp);
    }
Open stream.
<p>
<b>Response Errors Enum:</b>
<ul>
    <li>
        <code>STREAM_FENCED</code>:
         <ol>
             <li> stream's epoch is larger than request epoch </li>
             <li> stream's current range's node is not equal to request node </li>
             <li> stream's epoch matched, but stream's state is <code>CLOSED</code> </li>
         </ol>
    </li>
    <li>
        <code>STREAM_NOT_EXIST</code>
        <ol>
            <li> stream's id not exist in current stream-metadata </li>
        </ol>
    </li>
    <li>
        <code>STREAM_NOT_CLOSED</code>
        <ol>
            <li> request with higher epoch but stream's state is <code>OPENED</code> </li>
        </ol>
    </li>
    <li>
        <code>STREAM_INNER_ERROR</code>
        <ol>
            <li> stream's current range not exist when stream has been opened </li>
        </ol>
    </li>
</ul>
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java