public List<RangeMetadata> checkRemovableRanges() {
        NavigableMap<Long, S3StreamObject> objects = new TreeMap<>();
        streamObjects.forEach((objectId, object) -> objects.put(object.startOffset(), object));
        // Get the first continuous data range which stream objects covered.
        long startOffset = -1L;
        long endOffset = -1L;
        for (S3StreamObject object : objects.values()) {
            if (startOffset == -1L) {
                startOffset = object.startOffset();
                endOffset = object.endOffset();
                continue;
            }
            if (object.startOffset() != endOffset) {
                break;
            }
            endOffset = object.endOffset();
        }
        if (startOffset == -1L) {
            return Collections.emptyList();
        }
        List<RangeMetadata> removableRanges = new ArrayList<>();
        List<RangeMetadata> ranges = this.ranges.values().stream().sorted(RangeMetadata::compareTo).collect(Collectors.toList());
        for (int i = 0; i < ranges.size() - 1; i++) {
            // ranges.size() - 1: skip the last active range
            RangeMetadata range = ranges.get(i);
            if (startOffset <= range.startOffset() && range.endOffset() <= endOffset) {
                removableRanges.add(range);
            }
        }
        return removableRanges;
    }
Return the owner (node id) of current range.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamRuntimeMetadata.java