private ControllerResult<OpenStreamResponse> retryOpen(StreamRuntimeMetadata streamMetadata, long nodeId,
        long nodeEpoch,
        OpenStreamRequest req, OpenStreamResponse resp, AutoMQVersion version) {
        long streamId = req.streamId();
        long epoch = req.streamEpoch();
        // node may use the same epoch to open -> close -> open stream.
        // verify node
        RangeMetadata rangeMetadata = streamMetadata.ranges().get(streamMetadata.currentRangeIndex());
        if (rangeMetadata == null) {
            // should not happen
            log.error("[OpenStream] the current range not exist. streamId={}, streamEpoch={}, currentRangeIndex={}, nodeId={}, nodeEpoch={}",
                streamId, streamMetadata.currentEpoch(), streamMetadata.currentRangeIndex(), nodeId, nodeEpoch);
            resp.setErrorCode(Errors.STREAM_INNER_ERROR.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        if (rangeMetadata.nodeId() != nodeId) {
            log.warn("[OpenStream] the current range owner mismatch. streamId={}, streamEpoch={}, currentRangeIndex={}, ownerId={}, nodeId={}, nodeEpoch={}",
                streamId, streamMetadata.currentEpoch(), streamMetadata.currentRangeIndex(), rangeMetadata.nodeId(), nodeId, nodeEpoch);
            resp.setErrorCode(Errors.STREAM_FENCED.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // epoch equals, node equals, regard it as redundant open operation, just return success
        resp.setStartOffset(streamMetadata.startOffset());
        // Fix https://github.com/AutoMQ/automq/issues/1222
        // Handle the case: trim offset beyond the end offset
        resp.setNextOffset(Math.max(streamMetadata.startOffset(), streamMetadata.endOffset()));
        List<ApiMessageAndVersion> records = new ArrayList<>();
        if (streamMetadata.currentState() == StreamState.CLOSED) {
            records.add(new ApiMessageAndVersion(new S3StreamRecord()
                .setStreamId(streamMetadata.streamId())
                .setEpoch(epoch)
                .setRangeIndex(streamMetadata.currentRangeIndex())
                .setStartOffset(streamMetadata.startOffset())
                .setStreamState(StreamState.OPENED.toByte()), version.streamRecordVersion()));
        }
        return ControllerResult.of(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