public ControllerResult<CloseStreamResponse> closeStream(int nodeId, long nodeEpoch, CloseStreamRequest request) {
        CloseStreamResponse resp = new CloseStreamResponse();
        long streamId = request.streamId();
        long epoch = request.streamEpoch();

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

        // verify ownership
        Errors authResult = streamOwnershipCheck(streamId, epoch, nodeId, "CloseStream");
        if (authResult != Errors.NONE) {
            resp.setErrorCode(authResult.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamId);
        if (streamMetadata.currentState() == StreamState.CLOSED) {
            // regard it as a redundant close operation, just return success
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // now the request is valid, update the stream's state
        // stream update record
        List<ApiMessageAndVersion> records = List.of(
            new ApiMessageAndVersion(new S3StreamRecord()
                .setStreamId(streamId)
                .setEpoch(epoch)
                .setRangeIndex(streamMetadata.currentRangeIndex())
                .setStartOffset(streamMetadata.startOffset())
                .setStreamState(StreamState.CLOSED.toByte()),
                featureControlManager.autoMQVersion().streamRecordVersion()));
        log.info("[CloseStream] successfully close the stream. streamId={}, streamEpoch={}, nodeId={}, nodeEpoch={}",
            streamId, epoch, nodeId, nodeEpoch);
        return ControllerResult.atomicOf(records, resp);
    }
Close 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>
        </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_INNER_ERROR</code>
        <ol>
            <li> stream's current range not exist when stream has been opened </li>
            <li> close stream with higher epoch </li>
        </ol>
    </li>
</ul>
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java