public ControllerResult<DeleteStreamResponse> deleteStream(DeleteStreamRequest request) {
        DeleteStreamResponse resp = new DeleteStreamResponse();

        long streamId = request.streamId();

        StreamRuntimeMetadata streamMetadata = this.streamsMetadata.get(streamId);
        if (streamMetadata == null) {
            log.warn("[DELETE_STREAM],[FAIL]: stream not exist. streamId={}", streamId);
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        if (streamMetadata.currentState() != StreamState.CLOSED) {
            log.warn("[DELETE_STREAM],[FAIL]: stream is not closed. streamId={}", streamId);
            resp.setErrorCode(Errors.STREAM_NOT_CLOSED.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }

        // generate remove stream record
        List<ApiMessageAndVersion> records = new ArrayList<>();
        records.add(new ApiMessageAndVersion(new RemoveS3StreamRecord()
            .setStreamId(streamId), (short) 0));
        // generate stream objects destroy records
        List<Long> streamObjectIds = new ArrayList<>(streamMetadata.streamObjects().keySet());
        // deep delete the composite object: delete the composite object and it's linked objects
        ControllerResult<Boolean> markDestroyResult = this.s3ObjectControlManager.markDestroyObjects(streamObjectIds, Collections.nCopies(streamObjectIds.size(), CompactOperations.DEEP_DELETE));
        if (!markDestroyResult.response()) {
            log.error("[DELETE_STREAM],[FAIL]: failed to mark destroy stream objects. streamId={}, objects={}", streamId, streamObjectIds);
            resp.setErrorCode(Errors.STREAM_INNER_ERROR.code());
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        records.addAll(markDestroyResult.records());
        // the data in stream set object will be removed by compaction
        if (resp.errorCode() != Errors.NONE.code()) {
            return ControllerResult.of(Collections.emptyList(), resp);
        }
        log.info("[DELETE_STREAM]: successfully delete the stream. streamId={}", streamId);
        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