public int currentRangeOwner() {
        if (ranges.isEmpty()) {
            // there is no range in a new stream
            return -1;
        }
        return ranges.get(currentRangeIndex.get()).nodeId();
    }
Return the owner (node id) of current range.
metadata/src/main/java/org/apache/kafka/controller/stream/StreamRuntimeMetadata.java