ControllerResult<Boolean> checkStreamSetObjectExpired(S3StreamSetObject object,
        List<StreamOffsetRange> streamOffsetRanges) {
        boolean alive = false;
        for (StreamOffsetRange streamOffsetRange : streamOffsetRanges) {
            StreamRuntimeMetadata stream = streamsMetadata.get(streamOffsetRange.streamId());
            if (stream != null && stream.startOffset() < streamOffsetRange.endOffset()) {
                return ControllerResult.of(Collections.emptyList(), false);
            }
        }
        List<ApiMessageAndVersion> records = new ArrayList<>(2);
        records.add(new ApiMessageAndVersion(new RemoveStreamSetObjectRecord()
            .setNodeId(object.nodeId())
            .setObjectId(object.objectId()), (short) 0));
        records.addAll(this.s3ObjectControlManager.markDestroyObjects(List.of(object.objectId())).records());
        LOGGER.info("clean up scaled-in node={} object={}", object.nodeId(), object.objectId());
        return ControllerResult.of(records, true);
    }
Check whether this node is valid to operate the stream related resources.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamControlManager.java