public ControllerResult<TrimStreamResponse> trimStream(int nodeId, long nodeEpoch, TrimStreamRequest request) {
        long epoch = request.streamEpoch();
        long streamId = request.streamId();
        long newStartOffset = request.newStartOffset();
        TrimStreamResponse resp = new TrimStreamResponse();

        // verify node epoch
        Errors nodeEpochCheckResult = nodeEpochCheck(nodeId, nodeEpoch);
        if (nodeEpochCheckResult != Errors.NONE) {
            resp.setErrorCode(nodeEpochCheckResult.code());
            log.warn("[TrimStream] 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, "TrimStream");
        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) {
            log.warn("[TrimStream] can't trim a closed stream. streamId={}, streamEpoch={}, nodeId={}, nodeEpoch={}",
                streamId, epoch, nodeId, nodeEpoch);
            resp.setErrorCode(Errors.STREAM_NOT_OPENED.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        if (streamMetadata.startOffset() > newStartOffset) {
            log.warn("[TrimStream] trim offset less than start offset. streamId={}, streamEpoch={}, trimOffset={}, startOffset={}, nodeId={}, nodeEpoch={}",
                streamId, epoch, newStartOffset, streamMetadata.startOffset(), nodeId, nodeEpoch);
            resp.setErrorCode(Errors.OFFSET_NOT_MATCHED.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        if (streamMetadata.startOffset() == newStartOffset) {
            // regard it as a redundant trim operation, just return success
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // now the request is valid
        // update the stream metadata start offset
        List<ApiMessageAndVersion> records = new ArrayList<>();
        records.add(new ApiMessageAndVersion(new S3StreamRecord()
            .setStreamId(streamId)
            .setEpoch(epoch)
            .setRangeIndex(streamMetadata.currentRangeIndex())
            .setStartOffset(newStartOffset)
            .setStreamState(streamMetadata.currentState().toByte()), featureControlManager.autoMQVersion().streamRecordVersion()));
        // remove range or update range's start offset
        for (Map.Entry<Integer, RangeMetadata> entry : streamMetadata.ranges().entrySet()) {
            Integer rangeIndex = entry.getKey();
            RangeMetadata range = entry.getValue();
            if (newStartOffset <= range.startOffset()) {
                continue;
            }
            if (rangeIndex == streamMetadata.currentRangeIndex()) {
                // current range, update start offset
                // if current range is [50, 100)
                // 1. try to trim to 60, then the current range will be [60, 100)
                // 2. try to trim to 100, then the current range will be [100, 100)
                // 3. try to trim to 110, then current range will be [100, 100)
                long newRangeStartOffset = Math.min(newStartOffset, streamMetadata.endOffset());
                records.add(new ApiMessageAndVersion(new RangeRecord()
                    .setStreamId(streamId)
                    .setRangeIndex(rangeIndex)
                    .setNodeId(range.nodeId())
                    .setEpoch(range.epoch())
                    .setStartOffset(newRangeStartOffset)
                    .setEndOffset(streamMetadata.endOffset()), (short) 0));
                continue;
            }
            if (newStartOffset >= range.endOffset()) {
                // remove range
                records.add(new ApiMessageAndVersion(new RemoveRangeRecord()
                    .setStreamId(streamId)
                    .setRangeIndex(rangeIndex), (short) 0));
                continue;
            }
            // update range's start offset
            records.add(new ApiMessageAndVersion(new RangeRecord()
                .setStreamId(streamId)
                .setNodeId(range.nodeId())
                .setStartOffset(newStartOffset)
                .setEndOffset(range.endOffset())
                .setEpoch(range.epoch())
                .setRangeIndex(rangeIndex), (short) 0));
        }
        if (resp.errorCode() != Errors.NONE.code()) {
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        // the data in stream set object will be removed by compaction
        if (resp.errorCode() != Errors.NONE.code()) {
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        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